检验你的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)。
相关文章
|
机器学习/深度学习 存储 算法
深度学习中的稀疏注意力
深度学习中的稀疏注意力
633 0
|
16天前
|
人工智能 监控 数据挖掘
AiPy发布第五期大模型适配度测评报告:Claude、GLM、豆包位居前三,美团LongCat落后
10月13日,AiPy发布《大模型适配度测评第五期报告》,覆盖20款国内外主流大模型,聚焦数据分析、编程开发、UI设计等十大真实场景。报告从成功率、资源消耗、速度等多维度综合评估,Claude-Sonnet-4以90%成功率位居榜首,GLM-4.5、Doubao-Seed-1.6等国产模型表现亮眼,展现中国AI技术进步。测评发现代码质量、中文支持、任务规划仍是主要挑战,为用户选型与模型优化提供重要参考。
|
6月前
|
负载均衡 前端开发 Java
SpringCloud调用组件Feign
本文深入探讨微服务Spring体系中的Feign组件。Feign是一个声明式Web服务客户端,支持注解、编码器/解码器,与Spring MVC注解兼容,并集成Eureka、负载均衡等功能。文章详细介绍了SpringCloud整合Feign的步骤,包括依赖引入、客户端启用、接口创建及调用示例。同时,还涵盖了Feign的核心配置,如超时设置、拦截器实现(Basic认证与自定义)和日志级别调整。最后,总结了`@FeignClient`常用属性,帮助开发者更好地理解和使用Feign进行微服务间通信。
534 1
|
11月前
|
网络架构
|
11月前
|
移动开发 JavaScript 前端开发
html table+css实现可编辑表格的示例代码
html table+css实现可编辑表格的示例代码
332 12
快速确定网络号的范围
该文介绍了如何通过子网掩码快速确定IP地址中的网络号,避免复杂计算。以IP地址192.168.1.10和子网掩码255.255.255.0为例,网络号是前三个字节(192.168.1),加上子网掩码中主机号部分的0(.0),得到网络号192.168.1.0。此外,文中还提供了IP地址与子网掩码的二进制转换方法。
323 6
|
API Python
全面拥抱FastApi — 蓝图APIRouter
全面拥抱FastApi — 蓝图APIRouter
|
人工智能 供应链 安全
万字讲透:军工企业数字化转型转什么,如何做?
随着国防现代化目标的提出,军工行业景气度加速上升,企业纷纷扩产以满足新型装备加速列装的需求。航天科工集团的航天云网和中国电科的“数字电科”等项目展示了数字化转型的成效,如缩短研发周期、提高生产效率和降低成本。数字化转型对军工企业至关重要,能提升生产关系、增强竞争力,并实现生产制造和供应链的智能化。然而,转型面临挑战,包括传统认知边界、商业模式创新、技术合作共享、人才短缺和观念体制障碍。企业需制定数字化战略规划,重构组织与流程,加强网络安全,并确保人才和技术保障。案例显示,低代码平台如织信Informat可助力企业实现国产化、灵活的战略部署和数字化转型。
|
存储 SQL 测试技术
基于SpringBoot+Vue交通管理在线服务系统的开发(源码+部署说明+演示视频+源码介绍)(2)
基于SpringBoot+Vue交通管理在线服务系统的开发(源码+部署说明+演示视频+源码介绍)
123 0