检验你的fastapi掌握了吗

简介: 本内容系统讲解了 FastAPI 的核心功能与高级应用,包括路径参数定义、类型验证、Pydantic 模型、依赖注入、异步处理、权限校验、CORS 配置、错误处理、文档生成及性能优化等内容,适用于构建高效、可维护的现代 Web API 服务。

1. FastAPI 中接收路径参数的 GET 接口定义及类型验证实现

  • 接口定义方式:在路由装饰器(如@app.get)的 URL 路径中,用{参数名}声明路径参数,同时在视图函数参数中指定参数类型(如intstr)。
    示例: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 中的作用及请求体验证解析实现

  • 核心作用
  1. 请求体结构定义:明确接口接收的 JSON 请求体的字段、类型、必填 / 可选属性,让接口契约更清晰;
  2. 自动验证:对请求体字段进行类型、格式、范围等校验(如字符串长度、数字大小、邮箱格式);
  3. 数据解析与转换:将 HTTP 请求中的原始数据(如 JSON 字符串)自动解析为 Python 对象,同时将 Python 对象自动转换为响应 JSON;
  4. 文档自动生成:基于 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 依赖注入的典型场景及需释放资源的依赖项定义

  • 典型应用场景
  1. 资源共享:如数据库连接、Redis 客户端、HTTP 会话等,避免在每个接口中重复创建和销毁资源;
  2. 权限校验:如验证请求头中的令牌(Token)、检查用户是否登录 / 是否为管理员;
  3. 参数预处理:如从请求中提取用户 ID、解析分页参数(pagepage_size)并统一处理默认值;
  4. 配置加载:如加载环境变量、读取配置文件,供多个接口复用。
  • 需释放资源的依赖项定义(用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 调用),且有对应的异步库(如asyncpghttpx.AsyncClient);
2. 高并发场景,需最大化 CPU 利用率,减少 IO 阻塞带来的性能损耗
1. 接口中以 CPU 密集型操作为主(如数据计算、复杂逻辑处理),异步无法提升性能;
2. 无对应的异步库,只能使用同步库(如某些旧的数据库驱动);
3. 简单接口(如静态数据返回),同步实现更简洁


关键提醒:异步接口中不能直接调用同步阻塞函数(如requests.getmysql-connector的同步操作),否则会阻塞整个事件循环,导致所有异步接口无法响应;若必须使用同步函数,需通过BackgroundTasksThreadPoolExecutor将其放入后台线程执行。

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 通过两种方式自定义错误响应:


  1. 针对特定异常:用HTTPExceptiondetail参数自定义错误信息;
  2. 针对全局异常:用@app.exception_handler装饰器定义异常处理器,捕获特定异常(如404 Not Found500 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 规范)实现,核心逻辑:


  1. 接口信息收集:FastAPI 在启动时,会自动扫描所有带路由装饰器(@app.get/@app.post等)的视图函数,提取接口的路径、方法、参数(路径参数、查询参数、请求体)、响应类型等信息;
  2. OpenAPI schema 生成:将收集到的接口信息转换为符合 OpenAPI 规范的 JSON 数据(可通过http://localhost:8000/openapi.json查看);
  3. 文档 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)注意事项

  1. 参数必须可哈希lru_cache的 key 基于依赖函数的参数生成,因此参数必须是可哈希类型(如strinttuple),不能是listdict等不可哈希类型(否则会报错);
  2. 缓存失效问题lru_cache是内存缓存,且默认永久有效(除非服务重启或缓存达到maxsize后按 LRU 策略淘汰)。若依赖项的结果可能更新(如角色权限被修改),需手动清除缓存:
  • 方法:调用依赖函数的cache_clear()方法(如get_role_permissions.cache_clear()),通常在 “更新数据” 的接口中触发(如修改角色权限后,清除对应缓存);
  1. 避免缓存动态数据:不要缓存包含用户会话、实时数据(如实时库存)的依赖项,否则会导致不同用户获取到旧数据或他人数据;
  2. maxsize合理设置maxsize=None表示无限制缓存(不推荐,可能导致内存溢出),建议根据参数的可能取值数量设置maxsize(如角色类型只有 3 种,maxsize=10即可);
  3. 异步依赖不兼容lru_cache不能直接用于异步依赖函数(async def),若需缓存异步依赖,需使用第三方库(如functools.lru_cache的异步替代async_lru)。
相关文章
|
JSON 网络协议 机器人
ROSBridge简介以及理解使用(下)
ROSBridge简介以及理解使用(下)
2519 0
|
机器学习/深度学习 人工智能 搜索推荐
AIGC工具——文心一格
【1月更文挑战第12天】AIGC工具——文心一格
1306 3
AIGC工具——文心一格
|
4月前
|
缓存 API 数据库
Python性能优化利器:lru_cache装饰器详解
Python性能优化利器:lru_cache装饰器详解
|
iOS开发 Windows
思科网络模拟器7.3.1版本的下载和安装
思科网络模拟器7.3.1版本的下载和安装
2690 0
思科网络模拟器7.3.1版本的下载和安装
|
Shell iOS开发 MacOS
|
安全 物联网 网络安全
花无涯带你走进黑客世界9远控肉鸡(番外篇)
远程控制技术是黑客必学的技术之一,一台“肉鸡”的自述! 文章没有重复!!!,这是第九章的番外篇,“肉鸡”的自述!
503 0
|
2月前
|
SQL 算法 关系型数据库
击穿 InnoDB 事务隔离级别:RC 与 RR 的底层实现、锁机制、MVCC 与幻读终极拆解
本文深入剖析InnoDB事务隔离原理,聚焦RC(读已提交)与RR(可重复读)的核心差异:从锁机制(记录锁/间隙锁/临键锁)、MVCC可见性规则(Read View生成时机)到幻读解决方案。结合可复现实例与Java实战,助你彻底理解底层逻辑,规避90%的数据不一致与死锁问题。
308 3
|
Python
Flask三种文件下载方法
Flask 是一个流行的 Python Web 框架,它提供了多种方法来实现文件下载。在本文中,我们将介绍三种不同的方法,以便你能够选择最适合你应用程序的方法。
1401 2
|
人工智能 搜索推荐 算法
《AI赋能自由职业:开启竞争力提升新征程》
在数字化时代,AI为自由职业者带来巨大机遇。通过自动化任务处理、智能日程管理优化工作流程;借助AI工具提升写作、设计、翻译等专业技能和服务质量;利用数据分析和精准营销拓展业务与客户群体;并通过个性化学习路径和虚拟导师实现自我提升。积极拥抱AI,自由职业者能在竞争中脱颖而出,取得更大成功。
825 8
|
移动开发 JavaScript 前端开发
html table+css实现可编辑表格的示例代码
html table+css实现可编辑表格的示例代码
615 12