1. FastAPI 中接收路径参数的 GET 接口定义及类型验证实现
- 接口定义方式:在路由装饰器(如
@app.get)的 URL 路径中,用{参数名}声明路径参数,同时在视图函数参数中指定参数类型(如int、str)。
示例:python
from fastapi import FastAPI app = FastAPI() # 定义接收路径参数item_id(类型为int)的GET接口 @app.get("/items/{item_id}") def read_item(item_id: int): return {"item_id": item_id}
- 类型验证实现原理:FastAPI 底层依赖
Pydantic实现类型校验。当请求到达时,FastAPI 会自动将 URL 中的路径参数(默认是字符串格式)转换为视图函数指定的类型(如int);若转换失败(如传入非数字字符串给int类型参数),会直接返回422 Unprocessable Entity错误,并附带详细的验证失败信息(如"msg": "value is not a valid integer"),无需手动编写校验逻辑。
2. Pydantic 模型在 FastAPI 中的作用及请求体验证解析实现
- 核心作用:
- 请求体结构定义:明确接口接收的 JSON 请求体的字段、类型、必填 / 可选属性,让接口契约更清晰;
- 自动验证:对请求体字段进行类型、格式、范围等校验(如字符串长度、数字大小、邮箱格式);
- 数据解析与转换:将 HTTP 请求中的原始数据(如 JSON 字符串)自动解析为 Python 对象,同时将 Python 对象自动转换为响应 JSON;
- 文档自动生成:基于 Pydantic 模型的定义,自动在
/docs和/redoc文档中生成请求体的结构说明,无需手动编写文档。
- 请求体验证解析示例:python
from fastapi import FastAPI from pydantic import BaseModel, EmailStr # EmailStr是Pydantic提供的邮箱格式校验类型 app = FastAPI() # 定义Pydantic模型(即请求体结构) class UserCreate(BaseModel): username: str # 必填字段,默认校验类型为str email: EmailStr # 必填字段,校验邮箱格式 age: int | None = None # 可选字段,类型为int或None,默认值None # 视图函数接收Pydantic模型作为请求体参数 @app.post("/users") def create_user(user: UserCreate): # user已被自动解析为UserCreate对象,可直接通过属性访问字段 return {"username": user.username, "email": user.email, "age": user.age}
- 若请求体中
email字段格式错误(如"email": "invalid-email"),FastAPI 会返回422错误,提示"msg": "value is not a valid email address"。
3. FastAPI 依赖注入的典型场景及需释放资源的依赖项定义
- 典型应用场景:
- 资源共享:如数据库连接、Redis 客户端、HTTP 会话等,避免在每个接口中重复创建和销毁资源;
- 权限校验:如验证请求头中的令牌(Token)、检查用户是否登录 / 是否为管理员;
- 参数预处理:如从请求中提取用户 ID、解析分页参数(
page、page_size)并统一处理默认值; - 配置加载:如加载环境变量、读取配置文件,供多个接口复用。
- 需释放资源的依赖项定义(用
yield实现):
FastAPI 支持 “生成器依赖”,通过yield关键字拆分 “资源创建” 和 “资源释放” 逻辑:yield之前的代码用于创建资源(如建立数据库连接),yield返回的对象供视图函数使用,请求处理完成后,自动执行yield之后的代码释放资源(如关闭数据库连接)。
示例(数据库连接依赖):python
from fastapi import Depends import mysql.connector # 假设使用MySQL数据库 # 定义依赖项(生成器函数) def get_db(): # 1. 资源创建:建立数据库连接 db = mysql.connector.connect( host="localhost", user="root", password="password", database="test_db" ) try: yield db # 返回连接对象,供视图函数使用 finally: # 3. 资源释放:请求处理完成后,自动关闭连接 db.close() # 视图函数依赖注入数据库连接 @app.get("/items") def get_items(db=Depends(get_db)): # 2. 使用资源:通过db操作数据库 cursor = db.cursor(dictionary=True) cursor.execute("SELECT * FROM items") items = cursor.fetchall() return items
4. FastAPI 查询参数验证规则添加(举例说明)
FastAPI 通过Query类(从fastapi导入)为查询参数添加验证规则,支持长度限制、正则匹配、数值范围、必填 / 可选、默认值等校验。
示例(带多种验证规则的查询参数):
python
from fastapi import FastAPI, Query from typing import Optional app = FastAPI() @app.get("/search") def search( # 1. q:必填字符串,长度1-50,正则匹配“仅含字母/数字/下划线” q: str = Query( ..., # ... 表示必填(无默认值) min_length=1, max_length=50, regex=r"^[a-zA-Z0-9_]+$", description="搜索关键词(仅支持字母、数字、下划线,1-50字符)" ), # 2. page:可选整数,默认1,最小值1(避免页码为0或负数) page: Optional[int] = Query(1, ge=1, description="页码,默认1"), # 3. page_size:可选整数,默认10,范围5-100(限制单页数据量) page_size: Optional[int] = Query(10, ge=5, le=100, description="单页条数,5-100") ): return {"q": q, "page": page, "page_size": page_size}
- 若
q参数长度超过 50,会返回422错误:"msg": "ensure this value has at most 50 characters"; - 若
page参数传0,会返回422错误:"msg": "ensure this value is greater than or equal to 1"。
5. FastAPI 异步接口与同步接口的执行机制区别及适用场景
| 对比维度 | 异步接口(async def) |
同步接口(def) |
| 执行机制 | 基于 Python 异步 IO(async/await),接口执行过程中若遇到await(如异步数据库操作、异步 HTTP 请求),会 “让出” CPU,去处理其他请求,待await任务完成后再继续执行当前接口 |
基于同步阻塞模型,接口执行过程中会独占 CPU,遇到 IO 操作(如同步数据库查询、同步文件读写)时会阻塞,直到 IO 完成才释放 CPU |
| 依赖的运行环境 | 需搭配支持异步的 ASGI 服务器(如 Uvicorn、Hypercorn) | 可运行在 ASGI 服务器或 WSGI 服务器(如 Gunicorn),但在 ASGI 中会被放入线程池执行 |
| 适用场景 | 1. 接口中包含大量 IO 密集型操作(如数据库查询、Redis 操作、第三方 API 调用),且有对应的异步库(如asyncpg、httpx.AsyncClient);2. 高并发场景,需最大化 CPU 利用率,减少 IO 阻塞带来的性能损耗 |
1. 接口中以 CPU 密集型操作为主(如数据计算、复杂逻辑处理),异步无法提升性能; 2. 无对应的异步库,只能使用同步库(如某些旧的数据库驱动); 3. 简单接口(如静态数据返回),同步实现更简洁 |
关键提醒:异步接口中不能直接调用同步阻塞函数(如requests.get、mysql-connector的同步操作),否则会阻塞整个事件循环,导致所有异步接口无法响应;若必须使用同步函数,需通过BackgroundTasks或ThreadPoolExecutor将其放入后台线程执行。
6. 用依赖链实现多层级权限校验(如 “验证令牌→解析用户→检查管理员权限”)
依赖链的核心是 “依赖项本身依赖其他依赖项”,通过多层Depends嵌套,实现逐步递进的校验逻辑。以 “令牌验证→解析用户→管理员校验” 为例:
python
from fastapi import FastAPI, Depends, HTTPException from typing import Optional app = FastAPI() # 模拟:存储用户数据(实际项目中从数据库获取) fake_users = { 1: {"id": 1, "username": "admin", "is_admin": True}, 2: {"id": 2, "username": "user1", "is_admin": False} } # 第一层依赖:验证请求头中的Token(假设Token格式为“user_id:token_str”) def verify_token(token: Optional[str] = Depends(lambda: None)): # 简化:从请求头提取Token(实际用Header类更规范) if not token or ":" not in token: raise HTTPException(status_code=401, detail="无效的Token:Token不存在或格式错误") user_id_str, _ = token.split(":") if not user_id_str.isdigit(): raise HTTPException(status_code=401, detail="无效的Token:用户ID格式错误") return int(user_id_str) # 返回用户ID,供下一层依赖使用 # 第二层依赖:根据用户ID解析用户(依赖第一层的verify_token) def get_current_user(user_id: int = Depends(verify_token)): user = fake_users.get(user_id) if not user: raise HTTPException(status_code=401, detail="用户不存在") return user # 返回用户信息,供下一层依赖或视图函数使用 # 第三层依赖:检查用户是否为管理员(依赖第二层的get_current_user) def get_current_admin(user=Depends(get_current_user)): if not user["is_admin"]: raise HTTPException(status_code=403, detail="权限不足:仅管理员可访问") return user # 返回管理员信息,供视图函数使用 # 视图函数:依赖第三层的管理员校验(即触发完整依赖链) @app.get("/admin/items") def get_admin_items(admin=Depends(get_current_admin)): return {"msg": f"管理员{admin['username']}访问成功", "items": ["admin_item1", "admin_item2"]}
- 依赖链执行顺序:
verify_token(验 Token)→get_current_user(解析用户)→get_current_admin(验管理员权限); - 若某一层校验失败(如 Token 无效、用户不存在、非管理员),会直接返回对应 HTTP 错误,终止后续执行。
7. FastAPI 配置 CORS 跨域资源共享(允许特定域名)
FastAPI 通过CORSMiddleware(从fastapi.middleware.cors导入)配置 CORS,需指定允许的源(allow_origins)、请求方法(allow_methods)、请求头(allow_headers)等参数。
示例(允许特定域名跨域):
python
from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware app = FastAPI() # 配置CORS app.add_middleware( CORSMiddleware, allow_origins=["https://example.com", "http://localhost:3000"], # 允许的前端域名(精确匹配) # 若需允许所有域名,可设为 allow_origins=["*"](生产环境不推荐,建议精确指定) allow_credentials=True, # 允许跨域请求携带Cookie allow_methods=["*"], # 允许的HTTP方法(*表示所有,如GET、POST、PUT等) allow_headers=["*"], # 允许的请求头(*表示所有,如Authorization、Content-Type等) ) # 测试接口:前端https://example.com或http://localhost:3000可访问 @app.get("/api/data") def get_data(): return {"data": "跨域请求成功"}
- 关键参数说明:
allow_origins:最核心参数,指定允许跨域的前端域名列表(如["http://localhost:3000"]是本地 React/Vue 项目的默认地址);allow_credentials=True:若前端需要携带 Cookie(如登录态),必须设置为True,且allow_origins不能用"*"(需精确指定域名);allow_methods/allow_headers:默认值为["GET"]/[],生产环境建议根据实际接口需求缩小范围(如allow_methods=["GET", "POST"])。
8. FastAPI 自定义接口错误响应(如修改 404 错误格式)
FastAPI 通过两种方式自定义错误响应:
- 针对特定异常:用
HTTPException的detail参数自定义错误信息; - 针对全局异常:用
@app.exception_handler装饰器定义异常处理器,捕获特定异常(如404 Not Found、500 Internal Server Error)并返回自定义响应格式。
示例 1:自定义 404 错误响应
python
from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError, HTTPException app = FastAPI() # 1. 自定义404异常处理器(捕获“路径不存在”的异常) @app.exception_handler(404) async def custom_404_handler(request: Request, exc: HTTPException): return JSONResponse( status_code=404, content={ "code": 404, "msg": "自定义404提示:您访问的接口路径不存在", "request_url": str(request.url) # 附带请求的URL,方便排查 } ) # 2. 自定义请求参数验证错误(如422错误)的响应格式 @app.exception_handler(RequestValidationError) async def custom_validation_handler(request: Request, exc: RequestValidationError): return JSONResponse( status_code=422, content={ "code": 422, "msg": "请求参数验证失败", "errors": exc.errors(), # 保留原始错误详情 "request_url": str(request.url) } ) # 测试接口:访问不存在的路径(如http://localhost:8000/non-exist)会触发自定义404 @app.get("/items") def get_items(): return {"msg": "正常接口响应"}
- 访问
http://localhost:8000/non-exist时,返回的响应格式为:json
{ "code": 404, "msg": "自定义404提示:您访问的接口路径不存在", "request_url": "http://localhost:8000/non-exist" }
9. FastAPI 自动生成 API 文档的原理及隐藏接口的方法
(1)自动生成 API 文档的原理
FastAPI 的自动文档(/docs和/redoc)基于OpenAPI 规范(原 Swagger 规范)实现,核心逻辑:
- 接口信息收集:FastAPI 在启动时,会自动扫描所有带路由装饰器(
@app.get/@app.post等)的视图函数,提取接口的路径、方法、参数(路径参数、查询参数、请求体)、响应类型等信息; - OpenAPI schema 生成:将收集到的接口信息转换为符合 OpenAPI 规范的 JSON 数据(可通过
http://localhost:8000/openapi.json查看); - 文档 UI 渲染:
/docs:使用 Swagger UI 渲染 OpenAPI schema,支持在线调试接口(填写参数、发送请求、查看响应);/redoc:使用 ReDoc 渲染 OpenAPI schema,风格更简洁,适合展示接口文档(不支持在线调试)。
其中,参数的类型、验证规则、描述等信息,主要来自 Pydantic 模型定义和Query/Path/Body等类的参数配置。
(2)隐藏接口不被文档显示的方法
通过在路由装饰器中添加include_in_schema=False参数,可将接口排除在 OpenAPI schema 之外,从而不在/docs和/redoc中显示。
示例:
python
from fastapi import FastAPI app = FastAPI() # 正常显示在文档中的接口 @app.get("/public/items") def get_public_items(): return {"msg": "公开接口,显示在文档中"} # 隐藏的接口(include_in_schema=False) @app.get("/internal/secret", include_in_schema=False) def get_secret_data(): return {"msg": "内部接口,不显示在文档中"}
- 访问
http://localhost:8000/docs时,只能看到/public/items接口,/internal/secret接口不会出现; - 隐藏的接口仍可正常访问(如通过
curl或 Postman 调用),仅不对外暴露文档。
10. 结合 lru_cache 优化 FastAPI 频繁调用的依赖项及注意事项
(1)优化方法:用lru_cache缓存依赖项结果
对于频繁调用且返回结果相对稳定的依赖项(如查询数据库中的静态数据、解析配置文件),可在依赖函数上添加@lru_cache装饰器,缓存其返回结果,避免重复执行耗时操作(如重复查询数据库)。
示例(缓存数据库中 “角色权限” 的依赖项):
python
from fastapi import FastAPI, Depends from functools import lru_cache from typing import Dict app = FastAPI() # 模拟:从数据库查询角色权限(假设这是一个耗时操作) def _fetch_role_permissions_from_db(role: str) -> Dict[str, bool]: print(f"耗时操作:从数据库查询{role}的权限(仅第一次执行)") # 模拟数据库查询结果 permissions = { "admin": {"create": True, "read": True, "update": True, "delete": True}, "user": {"create": False, "read": True, "update": False, "delete": False} } return permissions.get(role, {"read": False}) # 用lru_cache缓存依赖项结果(key为role参数) @lru_cache(maxsize=128) # maxsize=128:最多缓存128个不同参数的结果 def get_role_permissions(role: str = Depends(lambda: "user")): # 简化:默认角色为user(实际从用户信息中获取) return _fetch_role_permissions_from_db(role) # 多个接口依赖get_role_permissions @app.get("/user/items") def get_user_items(permissions=Depends(get_role_permissions)): if permissions["read"]: return {"msg": "用户可读取items", "permissions": permissions} return {"msg": "无读取权限"} @app.get("/admin/items") def get_admin_items(permissions=Depends(lambda: get_role_permissions(role="admin"))): if permissions["delete"]: return {"msg": "管理员可删除items", "permissions": permissions} return {"msg": "无删除权限"}
- 首次调用
/user/items或/admin/items时,会执行_fetch_role_permissions_from_db并打印日志; - 后续调用时,直接从
lru_cache获取结果,不执行耗时的数据库查询,提升接口响应速度。
(2)注意事项
- 参数必须可哈希:
lru_cache的 key 基于依赖函数的参数生成,因此参数必须是可哈希类型(如str、int、tuple),不能是list、dict等不可哈希类型(否则会报错); - 缓存失效问题:
lru_cache是内存缓存,且默认永久有效(除非服务重启或缓存达到maxsize后按 LRU 策略淘汰)。若依赖项的结果可能更新(如角色权限被修改),需手动清除缓存:
- 方法:调用依赖函数的
cache_clear()方法(如get_role_permissions.cache_clear()),通常在 “更新数据” 的接口中触发(如修改角色权限后,清除对应缓存);
- 避免缓存动态数据:不要缓存包含用户会话、实时数据(如实时库存)的依赖项,否则会导致不同用户获取到旧数据或他人数据;
maxsize合理设置:maxsize=None表示无限制缓存(不推荐,可能导致内存溢出),建议根据参数的可能取值数量设置maxsize(如角色类型只有 3 种,maxsize=10即可);- 异步依赖不兼容:
lru_cache不能直接用于异步依赖函数(async def),若需缓存异步依赖,需使用第三方库(如functools.lru_cache的异步替代async_lru)。