fastapi【0基础学习之路(未学py版)】
第一部分:FastAPI 基础入门 (就像搭积木一样)
1.1 什么是 FastAPI?(给网站或App盖楼房的工具)
想象一下,你想做一个网站或者一个手机App,它们需要有一些后台的功能,比如用户可以注册登录,可以发布信息,可以查看数据等等。FastAPI 就像一个工具箱,里面有很多工具,可以帮助你快速地搭建起这些后台功能。
你可以把它想象成盖楼房,FastAPI 提供了各种“砖块”和“工具”,让你不用从头开始,就能很快地盖出一栋功能完善的“后台楼房”。这个“楼房”可以接收来自网站或App的信息,处理这些信息,然后再把结果告诉网站或App。
1.2 安装 FastAPI(准备好工具箱)
在我们开始使用这些工具之前,我们需要先准备好这个工具箱。在电脑上安装 FastAPI 和一个叫做 Uvicorn 的程序,就像你安装一个游戏或者其他软件一样。你需要打开一个叫做“终端”或者“命令提示符”的黑框框(在电脑上搜索一下就能找到),然后输入下面这行命令,让电脑自动帮你安装:
pip install fastapi uvicorn #有的时候因为py安装的原因无法使用 #可以用:py -m pip install fastapi uvicorn #或者: Python -m pip install fastapi uvicorn
你可以把它想象成从商店里购买 FastAPI 和 Uvicorn 这两个工具。
1.3 你的第一个 FastAPI 应用(搭一个最简单的房间)
现在我们来用 FastAPI 搭一个最简单的“房间”。创建一个叫做
main.py
的文件,用记事本或者其他可以写代码的软件打开,然后输入下面的代码:from fastapi import FastAPIapp = FastAPI()@app.get("/") async def read_root():return {"message": "你好,世界!"}
这段代码就像你用一些积木搭了一个只有一个房间的小房子。
from fastapi import FastAPI
: 这句话的意思是,我们从 FastAPI 这个工具箱里拿出盖房子需要的最基本的“工具”——FastAPI
。app = FastAPI()
: 这句话是说,我们用这个基本的“工具”创建了一个叫做app
的“房子”的地基。@app.get("/")
: 这就像给你的“房子”安装了一个前门,这个前门在网站地址的最后面,用/
表示(就像你访问一个网站的主页)。get
表示我们想通过这个门“获取”信息。async def read_root():
: 这定义了当你通过前门进来后,会发生什么。“你好,世界!” 就是我们会看到的信息。async def
是一个稍微高级的概念,你可以先简单理解为让电脑更高效地处理信息。return {"message": "你好,世界!"}
: 这表示电脑会返回一个消息,这个消息的内容是“你好,世界!”。{"message": "你好,世界!"}
是一种电脑可以理解的格式,叫做 JSON,它就像一个装有信息的盒子,盒子上写着 "message",里面装着 "你好,世界!"。要让这个小房子运行起来,你需要再次打开你的“终端”或“命令提示符”,进入到你保存
main.py
文件的那个文件夹里,然后输入这行命令:uvicorn main:app --reload
这就像给你的小房子通上电,让它可以工作。
现在,打开你的电脑上的浏览器(比如 Chrome 或者 Edge),在地址栏里输入
http://127.0.0.1:8000
,然后按下回车。你应该就能在浏览器上看到:
这说明你已经成功搭建了你的第一个 FastAPI 应用!
1.4 告诉电脑你想看什么信息(路径和参数)
刚才我们做的只是访问房子的“大门口”(
/
),现在我们来试试告诉电脑你想看房子里不同的“房间”或者想要一些特定的信息。路径参数: 想象一下你的房子里有不同的房间,你想告诉电脑你想去哪个房间。
修改
main.py
文件:from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}") async def read_item(item_id: int):return {"item_id": item_id}
这里,我们在
/items/
后面加了一个{item_id}
,就像给你的房子里的每个物品都贴上了一个标签。当你访问http://127.0.0.1:8000/items/123
时,电脑就会知道你想看标签是123
的那个物品的信息,然后read_item
函数就会把这个标签告诉你。int
的意思是告诉电脑,我们期望这里的item_id
是一个数字。在浏览器中尝试访问
http://127.0.0.1:8000/items/5
,你会看到:
查询参数: 有时候你想告诉电脑一些额外的信息,就像你在网上搜索的时候,会在搜索框里输入关键词。
修改
main.py
文件:from fastapi import FastAPIapp = FastAPI()@app.get("/items/") async def read_items(skip: int = 0, limit: int = 10):return {"skip": skip, "limit": limit}
当你访问
http://127.0.0.1:8000/items/?skip=2&limit=5
时,?
后面的skip=2
和limit=5
就是查询参数,它们告诉电脑你想要跳过前面 2 个物品,然后查看后面的 5 个物品。int = 0
和int = 10
表示这些参数是数字,并且有默认值。在浏览器中尝试访问
http://127.0.0.1:8000/items/?skip=1&limit=3
,你会看到:
1.5 给电脑一些信息(请求体)
有时候,你不仅想从电脑那里获取信息,还想给电脑发送一些信息,比如你注册一个账号的时候,需要告诉电脑你的用户名和密码。这就像你填写一个表格,然后交给电脑。
我们还是需要先定义一个“表格”的格式,在 FastAPI 中,我们用一种叫做 Pydantic 模型的东西来做这件事。
创建一个新的文件叫做
models.py
,然后输入以下内容:from pydantic import BaseModelclass Item(BaseModel):name: strprice: floatis_offer: bool = None
这个
Item
模型就像一个表格,它有三个格子:name
(名字,必须是文字)、price
(价格,必须是数字)、is_offer
(是否特价,可以是“是”或“不是”,也可以留空)。现在修改
main.py
文件:from fastapi import FastAPI from models import Itemapp = FastAPI()@app.post("/items/") async def create_item(item: Item):return item
这里,我们用了
@app.post("/items/")
,表示我们想通过/items/
这个“入口”发送信息给电脑。post
表示我们是“发送”信息。item: Item
表示我们期望发送的信息符合我们在models.py
里定义的Item
这个“表格”的格式。要发送信息,你不能直接在浏览器地址栏里输入网址了。你需要使用一些专门的工具,比如 Postman 或者 Insomnia。这些工具可以让你填写表格(也就是发送 JSON 格式的数据)然后发送给你的 FastAPI 应用。
假设你用 apifox 发送一个 POST 请求到
http://127.0.0.1:8000/items/
,请求体的内容是:
1.6 电脑告诉你结果(响应)
当我们发送请求给 FastAPI 应用后,它会处理我们的请求,然后返回一个结果,这个结果就是“响应”。就像我们刚才看到的
{"message": "你好,世界!"}
或者{"item_id": 5}
。
1.7 电脑会自动检查信息对不对(数据验证)
当我们定义了
Item
模型,并且告诉 FastAPI 我们期望接收Item
格式的数据时,FastAPI 会自动帮我们检查收到的数据是否符合要求。比如,如果我们发送的请求里name
不是文字,或者price
不是数字,FastAPI 会自动告诉我们哪里出错了,这就像电脑里的自动校对功能一样。
1.8 你有一个自动生成的说明书!(API 文档)
当你运行你的 FastAPI 应用后,你可以访问
http://127.0.0.1:8000/docs
,你会看到一个非常酷的页面。这个页面是 FastAPI 自动生成的 API 文档,它会列出你创建的所有“入口”(比如/
和/items/
),告诉你每个“入口”是用来做什么的,需要发送什么格式的信息,会返回什么格式的信息。这个文档就像你用 FastAPI 搭建的“后台楼房”的说明书,非常方便。
第二部分:FastAPI 进阶学习 (让你的楼房更高级)
2.1 更多指令 (除了看,还可以做其他事情)
在前面,我们学习了
@app.get("/")
用来获取信息,就像你访问一个网页,电脑会把网页的内容发给你。我们也学习了@app.post("/items/")
用来发送信息,就像你填写一个注册表格,把你的信息提交给网站。除了这两种操作,还有一些其他的“指令”,用来告诉电脑我们想对网站或App上的数据做不同的事情。这些“指令”叫做 HTTP 方法。
常见的 HTTP 方法有:
- GET: 用来从服务器获取信息。就像我们之前看到的,你告诉服务器你想看什么,服务器就把信息发给你。
- POST: 用来向服务器发送新的信息。就像你填写表格提交数据。
- PUT: 用来完整地更新服务器上已有的信息。想象一下你有一份文件,你想用一份新的完整的文件替换掉旧的文件。
- DELETE: 用来删除服务器上的信息。就像你从电脑里删除一个文件。
- PATCH: 用来部分地更新服务器上已有的信息。想象一下你有一份文件,你只想修改其中的一小部分内容,而不是完全替换掉。
我们来分别看看在 FastAPI 中怎么使用它们。
使用 PUT 方法更新信息
假设我们想更新一个已经存在的商品的信息。我们需要用到
@app.put()
。修改
main.py
文件,并确保models.py
文件还在:from fastapi import FastAPI from models import Itemapp = FastAPI()items = {1: {"name": "Old Product", "price": 10.0}, }@app.get("/items/{item_id}") async def read_item(item_id: int):if item_id in items:return items[item_id]return {"error": "Item not found"}@app.put("/items/{item_id}") async def update_item(item_id: int, item: Item):if item_id in items:items[item_id] = item.dict() # 将 Item 模型转换为字典存储return {"message": f"Item with ID {item_id} updated"}return {"error": "Item not found"}
- 我们首先创建了一个简单的字典
items
来存储一些商品信息,用商品的 ID 作为“标签”。@app.put("/items/{item_id}")
:这表示我们要更新 ID 为{item_id}
的商品信息。async def update_item(item_id: int, item: Item)
:这个函数接收商品的 ID 和新的商品信息(格式还是我们之前定义的Item
模型)。- 如果商品 ID 存在,我们就用新的商品信息替换掉旧的。
item.dict()
:这行代码把我们用Item
模型接收到的数据转换成电脑可以更容易存储的格式(一个字典)。要测试这个功能,你需要再次使用 Postman 或 Insomnia 这样的工具。发送一个 PUT 请求到
http://127.0.0.1:8000/items/1
,并在请求体中放入新的商品信息,比如:{"name": "New Product","price": 25.50 }
如果发送成功,你会收到类似
{"message": "Item with ID 1 updated"}
的响应。你也可以通过访问http://127.0.0.1:8000/items/1
来查看商品信息是否已经更新。
使用 DELETE 方法删除信息
假设我们想删除某个商品的信息。我们需要用到
@app.delete()
。继续修改
main.py
文件:from fastapi import FastAPI from models import Itemapp = FastAPI()items = {1: {"name": "Old Product", "price": 10.0}, }@app.get("/items/{item_id}") async def read_item(item_id: int):if item_id in items:return items[item_id]return {"error": "Item not found"}@app.put("/items/{item_id}") async def update_item(item_id: int, item: Item):if item_id in items:items[item_id] = item.dict()return {"message": f"Item with ID {item_id} updated"}return {"error": "Item not found"}@app.delete("/items/{item_id}") async def delete_item(item_id: int):if item_id in items:del items[item_id]return {"message": f"Item with ID {item_id} deleted"}return {"error": "Item not found"}
@app.delete("/items/{item_id}")
: 这表示我们要删除 ID 为{item_id}
的商品信息。async def delete_item(item_id: int)
: 这个函数接收要删除的商品的 ID。- 如果商品 ID 存在,我们就用
del items[item_id]
这行代码从我们的商品字典中删除这个商品的信息。要测试这个功能,你需要使用 apifox或 Insomnia 发送一个 DELETE 请求到
http://127.0.0.1:8000/items/1
。如果删除成功,你会收到类似{"message": "Item with ID 1 deleted"}
的响应。再次访问http://127.0.0.1:8000/items/1
,你应该会看到{"error": "Item not found"}
,表示商品已经被删除了。使用 PATCH 方法部分更新信息
假设我们只想更新某个商品的价格,而不想修改它的名字。我们可以使用
@app.patch()
。继续修改
main.py
文件,同时修改一下models.py
让name
和price
变成可选的:models.py:
from pydantic import BaseModel from typing import Optionalclass Item(BaseModel):name: Optional[str] = Noneprice: Optional[float] = Noneis_offer: Optional[bool] = None
main.py:
from fastapi import FastAPI from models import Itemapp = FastAPI()items = {1: {"name": "Old Product", "price": 10.0}, }@app.get("/items/{item_id}") async def read_item(item_id: int):if item_id in items:return items[item_id]return {"error": "Item not found"}@app.put("/items/{item_id}") async def update_item(item_id: int, item: Item):if item_id in items:items[item_id] = item.dict(exclude_unset=True) # 完整替换,忽略未设置的字段return {"message": f"Item with ID {item_id} updated"}return {"error": "Item not found"}@app.delete("/items/{item_id}") async def delete_item(item_id: int):if item_id in items:del items[item_id]return {"message": f"Item with ID {item_id} deleted"}return {"error": "Item not found"}@app.patch("/items/{item_id}") async def partial_update_item(item_id: int, item: Item):if item_id in items:existing_item = items[item_id]update_data = item.dict(exclude_unset=True) # 只包含设置了的字段updated_item = existing_item.copy()updated_item.update(update_data)items[item_id] = updated_itemreturn {"message": f"Item with ID {item_id} partially updated"}return {"error": "Item not found"}
- 在
models.py
中,我们使用了Optional
来表示name
和price
可以是空的。@app.patch("/items/{item_id}")
: 这表示我们要部分更新 ID 为{item_id}
的商品信息。async def partial_update_item(item_id: int, item: Item)
: 同样接收商品 ID 和Item
模型。item.dict(exclude_unset=True)
: 这行代码会把Item
模型转换成字典,但是只会包含我们设置了值的字段,没有设置值的字段会被忽略。- 我们先获取已有的商品信息,然后用我们想要更新的部分信息覆盖它。
要测试这个功能,使用 Postman 或 Insomnia 发送一个 PATCH 请求到
http://127.0.0.1:8000/items/1
,只在请求体中放入你想更新的部分,比如只更新价格:{"price": 30.0 }
发送成功后,再次访问
http://127.0.0.1:8000/items/1
,你会看到商品的名字还是原来的,但是价格已经更新为 30.0 了。总结
- GET 用来获取信息。
- POST 用来创建新的信息。
- PUT 用来完整地更新信息。
- DELETE 用来删除信息。
- PATCH 用来部分地更新信息。
这些不同的 HTTP 方法就像你和电脑之间的不同“指令”,告诉你电脑你想对数据做什么操作。
2.2 让电脑记住一些东西 (依赖注入)
我们刚才学习的是 2.1 更多指令 (HTTP 方法)。那么,我们现在开始学习 2.2 让电脑记住一些东西 (依赖注入)。
这个概念稍微有点抽象,但是我们可以用一个比喻来帮助理解。
想象一下你是一个厨师,你需要做一道菜。这道菜可能需要用到很多食材,比如鸡蛋、牛奶、面粉等等。
不使用依赖注入的情况:
每次你需要用到鸡蛋的时候,你都自己去冰箱里拿鸡蛋;每次需要用到牛奶的时候,你都自己去冰箱里拿牛奶。这样做当然可以,但是如果有很多菜都需要用到鸡蛋和牛奶,你就要重复很多次去冰箱拿的动作。
使用依赖注入的情况:
现在想象一下,有一个小助手在你旁边,你知道你需要哪些食材,你只需要告诉小助手你需要鸡蛋,小助手就会提前把鸡蛋准备好给你;你需要牛奶,小助手也会提前把牛奶准备好给你。这样你就不用每次都自己跑去冰箱拿了,你可以更专注于做菜。
在编程中,“食材”就是你的程序需要用到的各种功能或者数据,“小助手”就是 FastAPI 的依赖注入系统。它可以帮助你提前准备好你的程序需要的“食材”,让你在需要的时候直接使用,而不用每次都自己去“获取”。
在 FastAPI 中怎么使用依赖注入?
我们来看一个简单的例子。假设我们想知道用户在访问我们的网站时,他们的电脑是什么型号的浏览器。我们可以创建一个“小助手”函数来获取这个信息。
修改你的
main.py
文件:from fastapi import FastAPI, Header, Dependsapp = FastAPI()# 这是一个“小助手”函数,它可以获取用户浏览器的信息 async def get_user_agent(user_agent: str = Header(None)):return {"User-Agent": user_agent}@app.get("/") async def read_root(user_info: dict = Depends(get_user_agent)):return {"Hello": "World", "User Info": user_info}
让我们一步步解释一下:
from fastapi import FastAPI, Header
: 我们导入了FastAPI
和Header
。Header
用来告诉 FastAPI 我们想从请求的头部(Headers)中获取信息。async def get_user_agent(user_agent: str = Header(None)):
: 这是一个我们的“小助手”函数,它的名字是get_user_agent
。
user_agent: str = Header(None)
: 这表示我们想从请求的头部中获取一个叫做User-Agent
的信息。User-Agent
通常包含了用户浏览器的信息。Header(None)
告诉 FastAPI 从头部获取这个信息,如果找不到就给一个默认值None
。return {"User-Agent": user_agent}
: 这个函数会返回一个包含用户浏览器信息的字典。@app.get("/")
: 还是我们之前的根路径。async def read_root(user_info: dict = Depends(get_user_agent)):
: 这个函数是我们访问根路径时会执行的函数。
user_info: dict = Depends(get_user_agent)
: 这就是依赖注入的关键部分。Depends(get_user_agent)
的意思是告诉 FastAPI,在执行read_root
函数之前,先执行get_user_agent
这个“小助手”函数,然后把它的返回值(用户的浏览器信息)作为user_info
这个参数传递给read_root
函数。要测试这个,你可以直接在浏览器中访问
http://127.0.0.1:8000/
。你会看到类似这样的 JSON 响应:
注意
"User-Agent"
后面的那串文字,那就是你的浏览器告诉网站的它的信息。依赖注入的好处
就像我们的小助手可以提前准备好食材一样,依赖注入在编程中有很多好处:
- 代码更清晰: 你的主要代码(比如
read_root
函数)只需要关注核心逻辑,而不用自己去获取那些需要用到的信息或功能。- 更容易测试: 你可以很方便地替换掉“小助手”,用一些假的“食材”来测试你的主要代码是否工作正常。
- 代码可以重复使用: 你可以创建很多不同的“小助手”函数,然后在不同的地方使用它们。
现在你可能觉得这个概念有点新,没关系,你可以先记住这个比喻:依赖注入就像你身边有一个小助手,可以帮你准备好你需要的东西。在后面的学习中,你会慢慢体会到它的用处。
2.3 给你的网站或App加个门锁 (安全性)
想象一下你有一个自己的秘密小房间,你肯定不希望任何人都能随便进入。你需要一把锁,只有知道钥匙的人才能打开门。网站和App的安全性也是类似的,我们需要“门锁”来保护里面的信息,不让坏人随便查看或修改。
在网站和App的世界里,安全性主要涉及到两个方面:
- 身份验证 (Authentication): 证明“你是谁”:这就像你用钥匙打开你小房间的门。你需要提供一些信息来证明你的身份,比如你的用户名和密码。网站或App会检查你提供的信息是否正确,如果正确,就允许你进入。
- 授权 (Authorization): 证明“你能做什么”:就算你打开了门进入了房间,你也可能只能做某些事情,而不能做其他事情。比如,你可能可以看书,但是不能乱翻别人的东西。网站或App也会根据你的身份,决定你能访问哪些信息,能进行哪些操作。
我们先来看一个最简单的“门锁”—— HTTP 基本身份验证 (HTTP Basic Authentication)。虽然在实际应用中这种方式不太常用,因为它不是很安全,但是它可以帮助我们理解基本概念。
修改你的
main.py
文件:from fastapi import FastAPI, HTTPException, Depends from fastapi.security import HTTPBasic, HTTPBasicCredentialsapp = FastAPI()security = HTTPBasic()async def authenticate_user(credentials: HTTPBasicCredentials = Depends(security)):correct_username = "myusername"correct_password = "mypassword"if credentials.username == correct_username and credentials.password == correct_password:return credentials.usernameraise HTTPException(status_code=401, detail="Incorrect username or password")@app.get("/secret") async def read_secret_info(username: str = Depends(authenticate_user)):return {"message": f"Hello, {username}! This is secret information."}
我们来一步步解释一下:
from fastapi import FastAPI, HTTPException, Depends
: 导入我们需要的工具。HTTPException
用来在验证失败时返回错误信息。from fastapi.security import HTTPBasic, HTTPBasicCredentials
: 导入 HTTP 基本身份验证相关的工具。app = FastAPI()
: 创建我们的应用。security = HTTPBasic()
: 创建一个 HTTP 基本身份验证的“安全工具”。async def authenticate_user(credentials: HTTPBasicCredentials = Depends(security))
: 这是一个“小助手”函数,用来验证用户的身份。
credentials: HTTPBasicCredentials = Depends(security)
: 这表示我们希望通过 HTTP 基本身份验证的方式获取用户的凭证(用户名和密码)。Depends(security)
会让 FastAPI 自动处理从请求中提取用户名和密码的过程。correct_username = "myusername"
和correct_password = "mypassword"
: 这里我们设置了一个正确的用户名和密码,你可以想象成你小房间的钥匙。if credentials.username == correct_username and credentials.password == correct_password:
: 这行代码检查用户提供的用户名和密码是否和我们设置的正确值一样。return credentials.username
: 如果用户名和密码都正确,我们就返回用户名,表示用户通过了身份验证。raise HTTPException(status_code=401, detail="Incorrect username or password")
: 如果用户名或密码不正确,我们就告诉 FastAPI 返回一个错误信息,状态码是 401(表示“未授权”),错误内容是“Incorrect username or password”(用户名或密码错误)。@app.get("/secret")
: 我们创建了一个新的“入口”叫做/secret
,这个“入口”里的信息是我们想要保护起来的。async def read_secret_info(username: str = Depends(authenticate_user))
: 当我们访问/secret
这个“入口”时,会执行这个函数。
username: str = Depends(authenticate_user)
: 这表示在执行read_secret_info
函数之前,先执行authenticate_user
这个“小助手”函数来验证用户的身份。只有当authenticate_user
返回用户名时,read_secret_info
函数才会被执行,并且会接收到验证通过的用户名。return {"message": f"Hello, {username}! This is secret information."}
: 如果用户通过了身份验证,我们就会返回一条欢迎信息,并且包含一些秘密信息。现在,如果你尝试在浏览器中直接访问
http://127.0.0.1:8000/secret
,你会发现浏览器会弹出一个对话框,要求你输入用户名和密码。如果你输入
myusername
作为用户名,mypassword
作为密码,点击“确定”或“登录”,你就能看到 JSON 响应:如果你输入错误的用户名或密码,浏览器可能会显示一个错误页面,或者返回一个 401 错误。
总结
- 身份验证 (Authentication) 就是验证“你是谁”。在我们的例子中,就是检查你输入的用户名和密码是否正确。
- 授权 (Authorization) 就是验证“你能做什么”(虽然我们的例子里还没有很明显地区分授权,但是你可以想象,通过了身份验证的用户才能看到
/secret
里的内容)。这只是最简单的一种保护网站信息的方式。在实际应用中,安全性会更复杂,我们会使用更安全的方法,比如使用“令牌 (Tokens)” 等。但是理解了身份验证的基本概念,对你以后学习更高级的安全知识会有帮助。
2.4 让你的网站或App能存东西 (数据库集成 )
想象一下你的小房间里有很多重要的信息,比如你收集的邮票、你的日记、你喜欢的书等等。如果你只是把它们随便放在地上,时间久了就会很乱,找起来也很麻烦。你需要一个柜子或者一个书架,把它们分类整理好,这样你就能很容易地找到你需要的东西。
在网站和App的世界里,“柜子”或者“书架”就是数据库 (Database)。数据库是一个用来存储和管理数据的系统,它可以帮助我们把网站或App需要记住的信息(比如用户的账号信息、发布的帖子、商品的价格等等)有条理地存储起来,方便以后查找、修改和使用。
FastAPI 可以和很多不同种类的数据库“对话”,就像你可以有很多不同种类的柜子来存放你的东西。常见的数据库有:
- 关系型数据库 (Relational Databases): 就像一个有很多抽屉的柜子,每个抽屉里放着特定类型的信息,信息之间还有联系。比如 MySQL、PostgreSQL、SQLite。
- 非关系型数据库 (NoSQL Databases): 就像一个有很多不同大小和形状的盒子,你可以根据需要存放不同格式的信息。比如 MongoDB。
如何在 FastAPI 中使用数据库?
在 FastAPI 中连接和使用数据库,通常需要一些额外的工具和步骤。这部分内容相对来说比较复杂,对于初学者来说可能有点难。但是我们可以先简单了解一下基本思路。
一般来说,你需要做以下几件事情:
安装数据库驱动: 就像你需要特定的工具才能打开特定类型的柜子一样,你的 Python 程序也需要安装一些“驱动”,才能和特定的数据库“对话”。比如,如果你想用 MySQL 数据库,你需要安装一个叫做
mysqlclient
或者PyMySQL
的库。连接到数据库: 在你的 FastAPI 应用里,你需要编写代码来连接到你想要使用的数据库。这就像你走到你的柜子前面。
定义数据模型: 你需要告诉数据库你的数据是什么样的。这就像你给你的邮票、日记、书本分类,放到不同的抽屉里。在编程中,我们通常会用类似之前学习的 Pydantic 模型来定义数据库里的数据结构。
进行数据库操作: 你可以执行各种操作,比如:
- 创建数据 (Create): 往数据库里添加新的信息,就像你往柜子里放新的东西。
- 读取数据 (Read): 从数据库里查询信息,就像你从柜子里拿出你需要的东西。
- 更新数据 (Update): 修改数据库里已有的信息,就像你更换了书架上书的位置。
- 删除数据 (Delete): 从数据库里移除信息,就像你把不要的东西从柜子里扔掉。
一个简单的概念例子 (不包含实际代码):
from fastapi import FastAPI # 假设我们已经安装了连接数据库的工具app = FastAPI()# 假设我们已经连接到了一个叫做 "mydatabase" 的数据库# 假设我们有一个叫做 "items" 的表格,用来存储商品信息@app.get("/items_from_db") async def get_items_from_database():# 这里会写从数据库中读取所有商品信息的代码items = ... # 从数据库获取的商品信息return {"items": items}@app.post("/items_to_db/{name}/{price}") async def add_item_to_database(name: str, price: float):# 这里会写往数据库中添加新的商品信息的代码# 商品的名字是 {name},价格是 {price}... # 将商品信息存入数据库return {"message": f"商品 '{name}' 已添加到数据库"}
上面的代码只是一个很简单的例子,用来让你理解在 FastAPI 中使用数据库的大概思路。实际的代码会更复杂一些,涉及到如何编写查询语句、如何处理数据库返回的数据等等。
总结
- 数据库就像一个用来存储和管理信息的“柜子”。
- FastAPI 可以连接到各种不同类型的数据库。
- 使用数据库需要安装驱动、建立连接、定义数据模型和进行各种操作(创建、读取、更新、删除)。
数据库集成是构建功能完善的网站和App非常重要的一部分,但是它也相对复杂。我们现在只需要对它有一个基本的了解,知道它的作用是什么就可以了。如果你对这部分很感兴趣,我们以后可以再更深入地学习具体的技术。
2.5 像管道一样处理信息 (中间件)
好的,我们继续学习 2.5 像管道一样处理信息 (中间件)。
想象一下,你去一个大型的活动,比如一个演唱会。在你真正进入演唱会场地之前,你可能需要经过几个“关卡”:
- 安检: 检查你是否携带了违禁品。
- 验票: 检查你是否购买了门票。
- 指引: 工作人员告诉你座位在哪里。
这些“关卡”在你到达最终目的地(演唱会场地)之前,对你进行了一系列的处理。
在网站和App的世界里,中间件 (Middleware) 就有点像这些“关卡”。它是你发送给网站或App的请求(就像你走向演唱会场地)和网站或App返回给你的响应(就像你离开演唱会场地)之间的一层处理。你可以在中间件里做很多事情,比如:
- 检查请求: 看看请求是否符合某些规则(比如是否已经登录)。
- 修改请求: 在请求到达真正的处理程序之前,对请求进行一些修改。
- 记录日志: 记录下用户做了什么。
- 添加信息到响应: 在响应返回给用户之前,添加一些额外的信息。
如何在 FastAPI 中使用中间件?
在 FastAPI 中,你可以通过装饰器或者直接添加的方式来使用中间件。我们来看一个简单的例子,假设我们想记录下每个用户访问我们网站的时间。
修改你的
main.py
文件:from fastapi import FastAPI from time import timeapp = FastAPI()@app.middleware("http") async def add_process_time_header(request, call_next):start_time = time()response = await call_next(request) # 让请求继续传递给下一个环节(我们的路由函数)process_time = time() - start_timeresponse.headers["X-Process-Time"] = str(process_time) + " seconds"return response@app.get("/") async def read_root():return {"message": "Hello World"}@app.get("/items/{item_id}") async def read_item(item_id: int):return {"item_id": item_id}
让我们一步步解释一下:
from fastapi import FastAPI
: 导入 FastAPI。from time import time
: 导入time
模块,用于获取当前时间。app = FastAPI()
: 创建我们的应用。@app.middleware("http")
: 这是一个装饰器,用来告诉 FastAPI 我们要添加一个处理 HTTP 请求的中间件。async def add_process_time_header(request, call_next):
: 这是我们的中间件函数,它的名字是add_process_time_header
。
request
: 这个参数代表了用户发送给我们的请求的所有信息(比如他们访问的网址、他们发送的数据等等)。call_next
: 这是一个特殊的函数。你需要调用它来让请求继续传递给你的路由函数(比如read_root
或read_item
)。你的路由函数处理完请求后,会将响应返回给call_next
。start_time = time()
: 在请求被处理之前,我们记录下当前时间。response = await call_next(request)
: 我们调用call_next
函数,让请求继续传递给下一个环节。await
关键字是因为call_next
也是一个异步函数。call_next
会返回路由函数处理后的响应。process_time = time() - start_time
: 在响应返回之后,我们再次记录当前时间,然后计算出处理这个请求所花费的时间。response.headers["X-Process-Time"] = str(process_time) + " seconds"
: 我们在响应的头部(Headers)中添加了一个新的信息,叫做X-Process-Time
,它的值就是我们计算出来的处理时间。return response
: 最后,我们返回修改后的响应。现在,当你访问
http://127.0.0.1:8000/
或者http://127.0.0.1:8000/items/1
时,你会看到正常的 JSON 响应。但是,如果你查看响应的头部信息(在浏览器开发者工具的“Network”选项卡里可以看到),你会发现多了一个叫做X-Process-Time
的头部,它的值会告诉你处理这个请求花费了多少时间。总结
- 中间件就像请求和响应之间的“关卡”或“管道”。
- 你可以在中间件里对请求进行预处理,或者对响应进行后处理。
- 这可以帮助你做一些通用的事情,而不需要在每个路由函数里都写相同的代码。
2.6 允许其他人访问你的网站或App (CORS- 跨域资源共享)
想象一下你有两个不同的“在线商店”。你的第一个商店的网址是
mystore1.com
,你的第二个商店的网址是mystore2.com
。现在,假设你在mystore1.com
上展示了一些商品,并且你想在你的mystore2.com
上也显示这些商品的信息。问题来了,浏览器(你用来访问网站的程序,比如 Chrome 或 Edge)出于安全考虑,默认情况下不允许一个网站(比如
mystore2.com
)直接从另一个完全不同的网站(比如mystore1.com
)获取信息。这就像你的两个商店是独立的,一个商店里的顾客不能直接跑到另一个商店的后厨拿东西。这个安全规则叫做同源策略 (Same-Origin Policy)。“同源”指的是两个网站的协议 (http/https)、域名 (mystore1.com / mystore2.com) 和 端口号 (通常是 80 或 443) 都完全相同。如果其中任何一个不同,浏览器就会认为它们是不同的“来源”。
那么,如果你真的希望
mystore2.com
能够访问mystore1.com
的信息,该怎么办呢? 这就需要用到 跨域资源共享 (Cross-Origin Resource Sharing, CORS)。CORS 就像是
mystore1.com
这个商店给mystore2.com
发了一张“通行证”,允许mystore2.com
的顾客来我的商店后厨(也就是服务器)拿取一些特定的信息。如何在 FastAPI 中开启 CORS?
在 FastAPI 中,我们可以使用一个叫做
CORSMiddleware
的“中间件”来添加这个“通行证”。修改你的
main.py
文件:from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()origins = ["http://localhost:3000", # 允许这个网址的网站访问你的 API"https://your-other-website.com", # 允许这个网址的网站访问你的 API ]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"], # 允许所有 HTTP 方法 (GET, POST, PUT, DELETE, 等)allow_headers=["*"], # 允许所有请求头 )@app.get("/") async def read_root():return {"message": "Hello World"}@app.get("/items/{item_id}") async def read_item(item_id: int):return {"item_id": item_id}
让我们一步步解释一下:
from fastapi import FastAPI
: 导入 FastAPI。from fastapi.middleware.cors import CORSMiddleware
: 导入CORSMiddleware
,这是用来处理 CORS 的中间件。app = FastAPI()
: 创建我们的应用。origins = [...]
: 这里我们定义了一个列表,里面放着允许访问我们 API 的网站的网址。
"http://localhost:3000"
: 这通常是你本地开发前端代码时使用的网址。"https://your-other-website.com"
: 你可以把你想允许的其他网站的网址也加到这个列表里。app.add_middleware(...)
: 这行代码告诉 FastAPI 我们要添加一个中间件。
CORSMiddleware
: 我们要添加的是 CORS 中间件。allow_origins=origins
: 这个参数告诉 CORS 中间件,允许来自origins
列表中网址的请求访问我们的 API。你可以设置"*"
来允许所有来源的访问,但是通常不建议这样做,因为它不太安全。allow_credentials=True
: 这个参数表示是否允许发送 Cookie 等凭证信息(如果你的网站或App需要登录)。allow_methods=["*"]
: 这个参数表示允许哪些 HTTP 方法(比如 GET, POST, PUT, DELETE 等)进行跨域访问。"*"
表示允许所有方法。allow_headers=["*"]
: 这个参数表示允许哪些请求头进行跨域访问。"*"
表示允许所有请求头。通过添加这个中间件,你的 FastAPI 应用就会在响应中添加一些特殊的“头部信息”,告诉浏览器来自
origins
列表中的网站是被允许访问的。这样,你的mystore2.com
就可以安全地从你的 FastAPI 应用(运行在mystore1.com
或者localhost:8000
)获取商品信息了。总结
- 浏览器出于安全考虑,有同源策略,不允许跨域访问。
- CORS 是一种机制,允许服务器告诉浏览器,特定的其他来源是被允许访问它的资源的。
- 在 FastAPI 中,你可以使用
CORSMiddleware
来轻松地开启 CORS,并配置允许访问的来源、方法和头部。
2.7 让电脑偷偷帮你做一些事情 (后台任务)
有时候,当你在网站或App上做一些操作时,后台可能需要做一些额外的工作,但这些工作不是立刻就需要完成的,而且你也不需要一直等着它完成。
想象一下,你在一个在线商店买了一本书。当你完成付款后,你肯定希望立刻看到订单成功的页面。但是,商店的后台可能还需要做一些事情,比如:
- 发送一封确认邮件给你。
- 通知仓库打包你的书。
- 更新商品的库存数量。
这些事情都很重要,但是它们不需要在你点击“付款”按钮后立刻完成。让你一直等着这些事情都做完了再看到订单成功的页面,会感觉很慢。
后台任务 (Background Tasks) 就是用来处理这种不需要立刻完成的工作。你的网站或App可以先很快地响应你的操作,然后把这些后台任务交给电脑偷偷地去完成。
如何在 FastAPI 中使用后台任务?
在 FastAPI 中,你可以使用
BackgroundTasks
这个工具来添加后台任务。修改你的
main.py
文件:from fastapi import FastAPI, BackgroundTasks from time import sleepapp = FastAPI()def write_notification(email: str, message: str):print(f"发送邮件给 {email}: {message}")# 这里实际上会发送邮件,但是我们这里只是简单地打印出来sleep(5) # 模拟发送邮件需要一些时间@app.post("/send-notification/{email}") async def send_notification(email: str, background_tasks: BackgroundTasks):message = f"你好!感谢你的注册。"background_tasks.add_task(write_notification, email, message)return {"message": "通知已发送到后台"}
让我们一步步解释一下:
from fastapi import FastAPI, BackgroundTasks
: 导入FastAPI
和BackgroundTasks
。from time import sleep
: 导入sleep
函数,用来模拟耗时的操作。app = FastAPI()
: 创建我们的应用。def write_notification(email: str, message: str):
: 这是一个普通的 Python 函数,用来模拟发送邮件。它接收用户的邮箱和消息作为参数,然后简单地打印出来,并暂停 5 秒钟来模拟发送邮件需要的时间。@app.post("/send-notification/{email}")
: 我们创建了一个新的“入口”,当你向这个地址发送 POST 请求时,就会触发发送通知的操作。async def send_notification(email: str, background_tasks: BackgroundTasks):
: 这个函数处理发送通知的请求。
email: str
: 从路径中获取用户的邮箱。background_tasks: BackgroundTasks
: FastAPI 会自动把一个BackgroundTasks
对象传递给这个参数。我们可以使用这个对象来添加后台任务。message = f"你好!感谢你的注册。"
: 我们创建了一个简单的通知消息。background_tasks.add_task(write_notification, email, message)
: 这行代码就是关键!它告诉 FastAPI 在后台执行write_notification
这个函数,并将message
作为参数传递给它。注意,write_notification
函数并没有使用async
,它是一个普通的同步函数,FastAPI 也能在后台运行它。return {"message": "通知已发送到后台"}
: 这个函数会立刻返回一个消息,告诉你通知已经发送到后台了。你不需要等待邮件真的发送出去。现在,你可以使用 Postman 或 Insomnia 向
http://127.0.0.1:8000/send-notification/example@email.com
发送一个 POST 请求。你会立刻收到响应{"message": "通知已发送到后台"}
。同时,在你的终端(你运行uvicorn
的那个窗口)里,你会看到在 5 秒钟后打印出发送邮件给 example@email.com: 你好!感谢你的注册。
。这说明
write_notification
函数是在后台偷偷运行的,你的请求并没有因为发送邮件这个耗时的操作而等待。总结
- 后台任务用于处理不需要立刻完成的工作,让你的网站或App可以更快地响应用户的操作。
- 在 FastAPI 中,你可以使用
BackgroundTasks
来添加后台任务。- 你只需要定义一个普通的函数(或者异步函数),然后使用
background_tasks.add_task()
来告诉 FastAPI 在后台运行它。
2.8 检查你的房子结实不结实 (测试 )
想象一下你搭好了一个很棒的乐高玩具,或者你用纸板做了一个模型房子。你肯定希望确保你做的东西是牢固的,每个部分都按照你的想法工作。你会自己检查一下,看看有没有松动的地方,或者有没有哪个地方没有装对。
在编程中,“测试”就是做类似的事情。当我们写好了一段代码,比如我们的 FastAPI 应用,我们需要编写一些额外的代码来检查我们的应用是否工作正常,有没有 bug(错误)。这些额外的代码就是“测试”。
为什么要做测试?
- 确保代码正确: 测试可以帮助我们发现代码中的错误,确保我们的 API 按照预期工作。
- 防止未来出错: 当我们修改代码的时候,测试可以帮助我们确保新的修改不会破坏原有的功能。
- 提高代码质量: 好的测试可以让我们对自己的代码更有信心。
在 FastAPI 中怎么做测试?
在 Python 中,有一个很流行的工具叫做
pytest
,它可以帮助我们编写和运行测试。我们还需要 FastAPI 提供的一个叫做TestClient
的工具,它可以让我们模拟向我们的 FastAPI 应用发送请求。首先,你需要安装
pytest
:pip install pytest
然后,创建一个新的 Python 文件,通常我们把它叫做
test_main.py
,放在和你的main.py
文件同一个文件夹里。在这个文件里,我们可以编写我们的测试代码。假设我们要测试我们最开始创建的那个访问根路径
/
时返回"message": "Hello World"
的功能。你的test_main.py
文件可以这样写:from fastapi.testclient import TestClient from main import app # 导入你的 FastAPI 应用client = TestClient(app) # 创建一个测试客户端def test_read_main():response = client.get("/") # 模拟发送一个 GET 请求到根路径 "/"assert response.status_code == 200 # 检查返回的状态码是否是 200 (表示成功)assert response.json() == {"message": "Hello World"} # 检查返回的 JSON 内容是否和我们期望的一样
让我们一步步解释一下:
from fastapi.testclient import TestClient
: 我们从 FastAPI 导入TestClient
。from main import app
: 我们从我们的main.py
文件中导入我们创建的 FastAPI 应用实例app
。client = TestClient(app)
: 我们创建了一个TestClient
的实例,把它指向我们的 FastAPI 应用。这个客户端可以帮助我们模拟发送 HTTP 请求,而不需要真正运行我们的服务器。def test_read_main():
: 我们定义了一个测试函数,它的名字通常以test_
开头,这样pytest
才能找到它并运行。response = client.get("/")
: 我们使用client
的get()
方法模拟发送一个 GET 请求到根路径/
,这会得到一个response
对象,里面包含了服务器返回的所有信息。assert response.status_code == 200
:assert
是一个关键字,用来判断一个条件是否为真。这行代码检查服务器返回的状态码是否是 200。状态码 200 表示请求成功。如果返回的不是 200,测试就会失败。assert response.json() == {"message": "Hello World"}
: 这行代码检查服务器返回的 JSON 内容是否和我们期望的一样,也就是{"message": "Hello World"}
。如果不一样,测试也会失败。要运行这个测试,你需要打开你的终端,进入到包含
main.py
和test_main.py
文件的文件夹里,然后运行命令:pytest
如果你的测试代码写得正确,你会看到类似下面的输出,表示测试通过了:
.
表示一个测试通过了。如果测试失败了,你会看到F
或者其他的错误信息,告诉你哪个测试没有通过,以及为什么没有通过。你可以为你的每一个 API “入口”都编写类似的测试函数,来确保它们都工作正常。
总结
- 测试就像检查你的代码是否正确工作。
pytest
是一个常用的 Python 测试工具。- FastAPI 提供了
TestClient
来帮助你模拟发送 HTTP 请求并检查响应。- 你可以为你的每一个 API 功能都编写测试代码,确保它们都按照预期工作。
2.9 让你的网站或App更快 (异步编程)
你有没有试过在厨房里同时做很多事情?比如一边烧水,一边切菜,一边看烤箱里的蛋糕?如果你一件事情一件事情地做,可能会很慢。但是如果你能同时做几件事情,就能更快地完成所有的工作。
在电脑编程中,特别是当我们处理很多用户同时访问网站或App的时候,也需要让电脑能够“同时”处理很多任务,这样才能让大家都觉得很快。
同步编程 (Synchronous Programming) 就像你一件事情一件事情地做饭。如果你的程序需要等待某个操作完成(比如从数据库里读取数据,或者发送一个很慢的网络请求),那么在这段时间里,你的程序就什么都不能做,只能一直等待。这就像你在烧水的时候,就只能傻等着水开,不能同时切菜。如果有很多用户同时访问你的网站,每个人都要等待,体验就会很差。
异步编程 (Asynchronous Programming) 就像你可以同时做几件事情。当你的程序需要等待某个操作完成时,它不会傻等着,而是先去做其他的事情,等到原来的操作完成的时候,再回来处理结果。这就像你在烧水的时候,可以先去切菜,等水开了再回来处理水。这样你的程序就能更高效地利用时间,更快地响应用户的请求。
async
和await
是什么?在 Python 中,我们用两个关键词来做异步编程:
async
和await
。
async
: 用来定义一个异步函数 (asynchronous function)。就像我们在定义函数的时候用def
一样,我们定义异步函数的时候用async def
。异步函数有点特别,它可以在执行的过程中暂停,去做其他的事情,等到某个条件满足了再回来继续执行。
await
: 只能在async
函数里面使用。当你看到await
后面跟着一个“等待”的操作时(比如等待从网络上下载一个文件),你的程序就会暂停执行当前的异步函数,先去做其他的事情,等到这个“等待”的操作完成了,再回来继续执行。FastAPI 和异步编程
FastAPI 从一开始就被设计成支持异步编程的。你可能已经注意到了,我们之前定义的很多函数前面都有
async def
。这意味着 FastAPI 可以很高效地处理这些函数,即使里面有一些等待的操作。我们来看一个简单的例子:
from fastapi import FastAPI from time import sleepapp = FastAPI()def sync_task(seconds: int):print(f"同步任务开始,等待 {seconds} 秒...")sleep(seconds)print("同步任务结束!")return f"同步任务等待了 {seconds} 秒"async def async_task(seconds: int):print(f"异步任务开始,等待 {seconds} 秒...")await asyncio.sleep(seconds) # 注意这里我们用的是 asyncio.sleepprint("异步任务结束!")return f"异步任务等待了 {seconds} 秒"@app.get("/sync/{seconds}") async def run_sync(seconds: int):result = sync_task(seconds)return {"result": result}@app.get("/async/{seconds}") async def run_async(seconds: int):result = await async_task(seconds)return {"result": result}import asyncio
在这个例子中:
sync_task
是一个普通的同步函数,它会等待指定的秒数,期间什么都不做。async_task
是一个异步函数,它使用await asyncio.sleep(seconds)
来进行等待。asyncio
是 Python 中处理异步操作的一个库。await asyncio.sleep
不会阻塞整个程序,程序可以在等待的这段时间去做其他的事情。/sync/{seconds}
这个“入口”会运行同步任务。如果你访问/sync/5
,你的浏览器会等待 5 秒钟才能收到响应,因为在sync_task
函数执行期间,整个程序都在等待。/async/{seconds}
这个“入口”会运行异步任务。如果你访问/async/5
,你的浏览器可能不需要等待 5 秒钟才能收到响应(在真实的并发场景下才能更明显地看出区别),因为async_task
在等待的时候不会阻塞整个程序。总结
- 异步编程可以让你的程序更高效地处理多个任务,特别是在需要等待一些操作完成的时候。
async
用来定义异步函数,await
用来暂停异步函数的执行,等待某个操作完成。- FastAPI 天生支持异步编程,你可以使用
async def
来定义你的路由函数。
2.10 把你的房子放到网上 (部署)
你辛辛苦苦搭建好的“后台楼房”现在还只在你自己的电脑上运行。如果你想让其他人也能通过互联网访问你的网站或App,你需要把你的“房子”放到一个大家都能看到的地方。这个过程就叫做部署 (Deployment)。
你可以把部署想象成把你做好的乐高玩具模型或者纸板房子,从你自己的房间搬到一个公共的展览馆里,这样其他人就可以看到你的作品了。
在网站和App的世界里,“展览馆”就是一台服务器 (Server)。服务器是一种特殊的电脑,它一直开着,连接着互联网,可以用来运行你的网站或App,并响应用户的请求。
有很多不同的方法可以部署你的 FastAPI 应用,我们简单了解一下常见的几种方式:
使用云平台: 有一些公司提供了专门的服务,可以让你很方便地把你的应用部署到他们的服务器上,比如:
- Heroku: 一个非常容易上手 PaaS (Platform as a Service) 平台,很适合初学者。
- Render: 另一个简单易用的云平台,也支持部署 Python 应用。
- AWS (Amazon Web Services), Google Cloud, Azure: 这些是更强大也更复杂的云服务平台,提供了各种各样的服务,包括部署 Web 应用。
使用这些平台,你通常只需要把你的代码上传上去,然后进行一些简单的配置,平台就会帮你处理服务器的运行和维护。
使用 Docker: Docker 是一种工具,可以把你的整个应用和它所需要的运行环境打包在一起,就像把你的乐高模型装进一个盒子,这个盒子可以在任何支持 Docker 的服务器上运行。这使得部署更加方便和一致。
部署的基本步骤 (以一个简单的概念为主):
- 准备好你的代码: 确保你的所有 FastAPI 代码(比如
main.py
和models.py
文件)都准备好了。- 选择一个部署平台或方法: 根据你的需求和技术水平选择一个合适的平台或方法(比如 Heroku)。
- 配置你的应用: 你可能需要在部署平台上进行一些配置,比如告诉它你的应用的主文件是哪个(通常是
main.py
),以及如何运行你的应用(通常是使用uvicorn
)。- 上传你的代码: 将你的代码上传到你选择的部署平台。
- 让你的应用运行起来: 部署平台会启动你的服务器,并运行你的 FastAPI 应用。
- 你的应用现在可以在互联网上访问了! 你会得到一个公开的网址,其他人可以通过这个网址访问你的应用。
总结
- 部署就是把你的 FastAPI 应用放到互联网上的服务器上,让其他人可以访问。
- 服务器就像一个一直开着的电脑,连接着互联网。
- 有很多不同的方法可以部署 FastAPI 应用,比如使用云平台或者 Docker。
- 部署通常包括准备代码、选择平台、进行配置、上传代码和启动应用。
恭喜你!我们已经学习完了 FastAPI 进阶部分的所有主题。希望你对如何使用 FastAPI 构建 Web API 有了更深入的了解。