- 作者:老汪软件技巧
- 发表时间:2024-12-28 21:07
- 浏览量:
FastAPI 是一个现代、快速、非常易用的 Python Web 框架。它使用 Python 的类型提示(type hints)以及基于 Pydantic 的数据验证机制,极大地提升了开发效率和代码可维护性。
一、最新版本更新内容主要修复二、核心功能与使用示例1. 异步支持
FastAPI 原生支持异步编程模型,使用 Python 的 async/await 关键字即可实现高并发处理。
示例:
from fastapi import FastAPI
from typing import Union
app = FastAPI()
# 假设这是一个异步函数,用于模拟耗时的 I/O 操作
async def some_async_operation(item_id: int):
# 这里可以是对数据库的异步请求、网络请求等
return f"Processed item {item_id}"
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Union[str, None] = None):
# 异步处理请求
result = await some_async_operation(item_id)
return {
"item_id": item_id,
"q": q,
"result": result
}
使用要点:
在 Python 3.7+ 中,使用 async 和 await 关键字即可开启异步函数。配合异步服务器(如 Uvicorn 或 Hypercorn)一起运行,能发挥 FastAPI 高并发的优势。2. 自动数据验证
FastAPI 内部基于 Pydantic 进行模型校验。只需定义一个 Pydantic 模型,就可以在请求时自动完成参数的解析和验证,简化大量手动校验的工作。
示例:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# 定义 Pydantic 模型
class Item(BaseModel):
name: str
price: float
is_offer: bool = None
@app.post("/items/")
async def create_item(item: Item):
# 自动解析并验证请求体中的 JSON 数据
return {
"item_name": item.name,
"price": item.price,
"is_offer": item.is_offer
}
使用要点:
只需要在函数参数中加入模型类,就能自动完成请求体(JSON 或表单)的反序列化和校验。如果验证失败,FastAPI 会自动返回 422 错误(可在响应体中查看具体错误信息)。3. 依赖注入系统
FastAPI 拥有强大的依赖注入机制(Dependency Injection),可以让我们在路由函数中非常灵活地获取数据库连接、缓存对象、用户认证等资源,而不用手动在每个路由中反复声明。
示例:
from fastapi import Depends, FastAPI
app = FastAPI()
# 定义一个依赖函数
async def get_db_connection():
# 在这里实现数据库连接的获取和释放逻辑
return "db_connection"
@app.get("/users/")
async def read_users(db = Depends(get_db_connection)):
# db 参数即获取到的 "db_connection" 对象
return {"db": db}
使用要点:
通过 Depends() 引用依赖函数,FastAPI 会自动管理依赖函数的调用。依赖函数也可以互相依赖,从而形成可复用和分层的逻辑结构。在大型项目中,依赖注入可以让核心业务代码与基础设施管理解耦,提升可维护性。三、完整使用流程示例
以下简要展示如何从零开始创建并运行一个最简 FastAPI 项目,包含依赖注入、数据验证和异步操作等关键功能。
1. 安装 FastAPI
你可以使用 pip 来安装 FastAPI 及其常用依赖:
pip install "fastapi[standard]" uvicorn
说明:[standard] 表示安装 FastAPI 依赖的一些常用包,如 Pydantic、Starlette 等。
uvicorn 是一个常用的高性能异步服务器,用于运行 FastAPI 应用。
2. 创建项目文件
在你的项目文件夹中新建一个 main.py 文件,示例内容如下:
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import Union
app = FastAPI()
# 依赖函数:模拟数据库连接
async def get_db():
return "我是数据库连接示例"
# 定义 Pydantic 模型
class Item(BaseModel):
name: str
price: float
is_offer: bool = False
# 异步函数,用于模拟异步操作
async def async_process(item_id: int):
return f"异步处理完成:{item_id}"
@app.get("/items/{item_id}")
async def get_item(item_id: int, q: Union[str, None] = None):
result = await async_process(item_id)
return {
"item_id": item_id,
"query_param": q,
"async_result": result
}
@app.post("/items/")
async def create_item(item: Item, db=Depends(get_db)):
# 自动解析并验证请求体中的 JSON 数据
return {
"db_info": db,
"name": item.name,
"price": item.price,
"is_offer": item.is_offer
}
3. 启动服务
在命令行执行以下命令,即可启动 FastAPI 应用:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
启动成功后,在浏览器中访问:
四、性能优化建议1. 中间件(Middleware)的使用
为了在请求处理前后进行额外操作(如日志记录、性能监控等),可以编写自定义中间件,并通过 add_middleware 方法注册到应用中。
示例:
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class CustomHeaderMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# 在请求处理前执行操作
response = await call_next(request)
# 在返回响应后执行操作
response.headers["X-Custom-Header"] = "FastAPI"
return response
app.add_middleware(CustomHeaderMiddleware)
2. 使用异步数据库、缓存等3. 其他优化五、版本控制建议1. 固定 FastAPI 版本
在项目的依赖文件(如 requirements.txt 或 pyproject.toml)中指定 FastAPI 版本范围,以保证兼容性并获得必要的 bug 修复。
fastapi[standard]>=0.115.0,<0.116.0
这样能够: