基础

常用名称前浅析

常用工具/技术浅析

• 参数校验库合集
• Pydantic 是一个基于python类型提示来定义数据验证，序列化和文档(使用json模式)的库。一般是用在 定义数据模型 从而实现自动对数据数据进行验证；前端与后端交互传递数据的时候，可以通过Pydantic来验证传的数据是否符合规范， 返回给前端的时候也需要按照规范返回;
• Beanie 是一个现代化、高性能的 Python 对象 - 文档映射（Object-Document Mapping, ODM）库，主要用于和 MongoDB 数据库进行交互 1 4。它基于 Pydantic 构建，核心特性如下：
• 模型定义简单 ：通过继承 beanie.models.Document 类就能轻松定义数据模型，语法比传统 ORM 库（如 SQLAlchemy）更简洁。 #后端 #FastAPI #Python
• 异步支持 ：支持 Python 3.7+ 的异步编程，可在 awaitable 函数里执行数据库操作，提升应用整体响应速度。
• 内置 CRUD 操作 ：提供 create 、 get 、 update 、 delete 等方法，无需编写复杂查询语句。
• 实时变化追踪 ：能自动跟踪对象变化，并在需要时更新数据库，减少手动管理状态的复杂性。
• Starlette 是一个轻量级的 ASGI 框架，专注于提供 高性能 和简单的 API设计 。它提供了 路由 、 中间件 、 请求和响应处理 等基本功能，同时支持 异步编程 ，非常适合构建高性能的 Web 应用和 API，FastAPI是在Starlette 的基础上构建的
• WTForms：支持多个Web框架的Form组件，主要用于对用户请求数据进行校验验证
• validdr：轻量级、可拓展的数据验证和适配库
• validators：验证库
• cerberus：基于Python的轻量级和可拓展的数据验证库
• 。。。。。。
• uvicorn 参数说明
• 常用的类型注释
• Optional 是一个类型提示，用于表示一个变量可以是某种类型或者是 None
• List[T] : 表示一个列表，其中 T 可以是任何类型。例如， List[int] 表示一个整数列表。
• Dict[K, V] : 表示一个字典，其中 K 是键的类型， V 是值的类型。例如， Dict[str, int] 表示一个字典，键是字符串，值是整数。
• Tuple[T1, T2, ..., TN] : 表示一个元组，其中 T1 是第一个元素的类型， T2 是第二个元素的类型，等等。例如， Tuple[str, int] 表示一个包含一个字符串和一个整数的元组。
• Union[T1, T2, ..., TN] : 表示一个可以接受多种类型的参数，例如， Union[int, str] 表示一个可以是整数或字符串的参数。

逐个击破

Pydantic基础

简单理解就是规范请求体力的每个参数，可以提前指定好类型以及添加一些方法进行校验等，保证数据不会出错；

• 基本功能
• 使用 Python的类型注解来进行数据校验 和settings管理
• Pydantic可以在代码运行时提供类型提示，数据校验失败时提供友好的错误提示
• 定义数据应该如何在纯规范的Python代码中保存，并用Pydantic验证它

高级技巧

路由管理

一个应用里面会有很多子应用，主程序里是Fastapi类进行实例化，子应用是通过接口路由APPRouter的方式进行实例化，然后从主程序里面进行导入；因此可以通过相关的库进行应用和路由的管理

# tutorial/__init__.py# !/usr/bin/python3# -*- coding:utf-8 -*-# __author__ = '__Jack__'from .chapter03 import app03 from .chapter04 import app04 from .chapter05 import app05 from .chapter06 import app06 from .chapter07 import app07 from .chapter08 import app08

import uvicornfrom fastapi import FastAPI# tutorial 下面的每个py文件相当于一个应用，但是不能每一个都给它建立一个fastapi应用，所以这里通过接口路由的方式去实例化应用， 相当于子应用# 之所以这里能直接导入app03, 是因为在tutorial的__init__文件中导入app03了，就不用from tutorial.chapter03 import app03了from tutorial import app03, app04, app05# 示例化一个fastapi应用app = FastAPI()# 把接口路由的子应用接到主应用里面来# 这个前缀就是请求的url, tags表示应用的标题， api文档里面的接口上面都有标题名app.include_router(app03, prefix='/chapter03', tags=['第三章 请求参数和验证']) app.include_router(app04, prefix='/chapter04', tags=['第四章 响应处理和 FastAPI 配置']) app.include_router(app05, prefix='/chapter05', tags=['第五章 FastAPI 的依赖注入系统'])# 这里也可以只设置应用入口， 具体的prefix， 以及tags在相应的应用里面设置， 这个在实际开发中，会降低主应用与子应用的耦合性# 主应用里面只管导入子应用， 不管子应用的路径以及tagsapp.include_router(app03)if __name__ == "__main__": # 等价于之前的命令行启动：uvicorn run:app --reloaduvicorn.run("run:app", host='0.0.0.0', port=8000, reload=True, debug=True, workers=1)

所谓路由注册就是提供一个对应的URL地址来关联或绑定定义的函数，通常是使用 装饰器 的形式对需要绑定的函数进行映射绑定；

使用装饰器装饰的函数可能有两种，一种是使用def定义的 同步函数 ，另一种是使用async def定义的 协程函数 ；同步函数(基于 多线程并发模式 进行处理的)会运行于 外部的线程池 中，协程函数(基于 单线程内的异步并发模式 进行处理的)会运行于 异步事件循环 中;

• 同步路由
• 当使用URL地址绑定的关联视图函数是一个同步函数(使用def定义的函数)，则是同步路由
• 同步路由的并发处理机制是基于多线程方式实现的 → 不同线程ID
• 异步路由
• 使用URL地址绑定的视图函数是一个协程函数(使用async def定义的函数)时
• 同异步路由的并发处理机制是基于单线程方式实现的 → 同一个线程ID
• 一般不建议在异步操作中使用同步函数，因为一个线程中执行同步函数时会引发阻塞，如使用 await asyncio.sleep(10) 而不是直接使用 time.sleep(10)

@app.get(path="/async")async def asyncdef():    await asyncio.sleep(10) // 异步执行 不会阻塞同步执行逻辑    # time.sleep(10) // 同步的方式 会阻塞后续操作print("当前协程运行的线程 ID:", threading.current_thread().ident) return {"index": "async"}

在fastapi中，所有的路由都会统一保存到app.routers中，可以在应用中通过app.routers拿到所有注册的路由信息；内部会有实例和装饰器的存在，如app是fastapi的实例，而@app是实例app的装饰器，装饰器可以简化路由定义、解耦业务逻辑和路由配置

参数管理(path、Query、Body、dict、Field)

Fastapi中的参数有查询参数、路径参数、默认参数、可选参数、混合所有类型的参数；也可以结合 pydantic 模块创建 模型类 或 嵌套模型 等

混合所有类型的参数示例

# 首先应该声明所有必须的参数，然后是默认参数，最后是可选参数@app.patch("/ch01/account/profile/update/names/{username}") def update_profile_names(id: UUID, username: str = '', new_names: Optional[Dict[str, str]] = None): """ 处理更新用户姓名的请求。如果用户不存在，返回用户不存在消息。如果新姓名信息为空，返回需要新姓名的消息。否则，验证 ID 是否匹配，若匹配则更新用户个人资料中的姓名信息，返回更新成功的消息。否则返回用户不存在消息。""" if valid_users.get(username) is None: return {"message": "user does not exist"} elif new_names is None: return {"message": "new names are required"} else: user = valid_users.get(username) if user.id == id: profile = valid_profiles[username] profile.firstname = new_names['fname'] profile.lastname = new_names['lname'] profile.middle_initial = new_names['mi'] valid_profiles[username] = profile return {"message": "successfully updated"} else: return {"message": "user does not exist"}

• 主要有如下参数类型
• 路径相关参数
• Query参数
• Path参数
• Body参数
• Field参数
• 上述参数类型的异同点
• 相同点
• 进行参数验证和类型检查：
• 增强代码可读性和可维护性：
• 不同点
• Field：主要用于定义数据模型内部的字段定义，而不是直接处理请求参数，可以通过Body等工具结合使用
• Body：
• 主要用于 需要携带请求体 的请求方式
• 用于从请求体中获取数据，在制定的方法中可以帮助我们提取和处理请求体中的数据
• Path：用于提取URL中的参数，并就那些类型检查和验证
• Query：用于提取和处理URL中的查询参数
• 通常用于 GET 请求，因为 GET 请求通常通过查询字符串传递参数。但在其他请求方式中，也可以使用查询字符串传递额外的参数，此时同样可以使用 Query 来处理
• 路径操作 参数和 路径函数 参数
• 在配置API端点路由绑定视图函数的过程中会涉及到路径操作参数和路径函数参数，如 @app.post() 就是API端点的路径操作即API装饰器，内部的参数可以理解为 路径操作参数 ，紧跟着的 def/async def函数 即表示 视图函数 ，其中需要传入的参数表示路径函数参数即视图函数参数；

路径操作参数中的变量fastapi会自动将这两个参数传递到视图函数上；一般路径参数中不会直接以 / 关键字路径参数结尾，一般是 @app.get("/urls/{file_path:path}") 会自动识别出多重路径，后面的 path 表示该参数匹配任意路径，可以理解为 路径的转换机制 ，当然也可以通过 枚举预设路径参数值 来实现路径参数的定义

@app.get("/uls/{file_path:path}")async def callback_file_path_2(file_path: str):    return {        'file_path': file_path    }class ModelName(str, Enum): name1 = "name1" name2 = "name2" name3 = "name3"# 接收一个路径参数`model_name`，这个参数是一个枚举类型，有三个可能的值："name1"，"name2"，"name3"。这个接口的响应取决于`model_name`的值@app.get("/model/{model_name}") async def get_model(model_name: ModelName): if model_name == ModelName.name1: return {"model_name": model_name, "message": "ok!"} if model_name.value == "name2": return {"model_name": model_name, "message": "name2 ok!"} return {"model_name": model_name, "message": "fail!"}@app.get("/pay/{user_id}/article/{article_id}") async def callback(user_id: int = Path(..., title="用户 ID", description='用户 ID 信息', ge=10000), article_id: str = Path(..., title="文章 ID", description='用户所属文章 ID 信息', min_length=1, max_length=50)): return { 'user_id': user_id, 'article_id': article_id }

• Query参数

主要体现在 视图函数 中， Query逻辑 和 Path参数 逻辑大致相同，可以进行查询参数的多维度条件限制

@app.get("/query/")async def callback(user_id: int, user_name: Optional[str] = None, user_token: str = 'token'):    return {        'user_id': user_id,        'user_name': user_name,        'user_token': user_token    }@app.get("/query/bool/") async def callback(isbool: bool = False): return { 'isbool': isbool }@app.get("/query/morequery") async def callback( user_id: int = Query(..., ge=10, le=100), user_name: str = Query(None, min_length=1, max_length=50, regex="^fixedquery$"), user_token: str = Query(default='token', min_length=1, max_length=50), ): return { 'user_id': user_id, 'user_name': user_name, 'user_token': user_token }@app.get("/query/list/") async def query_list(q: List[str] = Query(["test1", "test2"])): return { 'q': q }

• Body(请求体)参数
• 接受Body参数的方式
• 引入Pydantic类型来生命请求体并进行绑定

from pydantic import BaseModelfrom typing import Optionalclass Item(BaseModel): user_id: str token: str timestamp: str article_id: Optional[str] = None@app.post("/action/") def callback(item: Item): return { 'user_id': item.user_id, 'article_id': item.article_id, 'token': item.token, 'timestamp': item.timestamp }

• 通过Request对象获取body的函数

from fastapi import FastAPI, Query, Path, Body@app.post("/action/body")def callbackbody(        token: str = Body(...),        user_id: int = Body(..., gt=10),        timestamp: str = Body(...),        article_id: str = Body(default=None),):    return {        'user_id': user_id,        'article_id': article_id,        'token': token,        'timestamp': timestamp    }

• 使用Body类来定义，可进行嵌套

class User(BaseModel):    username: str    full_name: str = Noneclass ItemUser3(BaseModel): name: str description: str = None price: float tax: float = None user: User # 新增模型嵌套并设置为集合类型tags: Set[str] = [] users: List[User] = None@app.put("/items/body5") async def update_item(item: ItemUser3, importance: int = Body(..., gt=0)): results = {"item": item, "user": item.user, "importance": importance} return results

• 通过dict字典类型构成请求体

@app.put("/demo/dict/{id:int}")async def update_item(id:int, item: Dict[str,str], user: Dict[str,str], gornd:Dict[str,Dict[str,int]]):    results = {"item": item, "user": user, "gornd": gornd}    return results

• 通过Pydantic中的Field

class field_demo(BaseModel):    type: str    address: list[list[str | int]]    title: list[str | int]    name: str = Field(..., min_length=3, max_length=10, title="用户名")    age: int = Field(..., ge=18, le=100, description="用户年龄")@app.post("/demo/field")async def update_item(body: field_demo, id:int, name: str = Query(None, min_length=1, max_length=50, regex="^fixedquery$")):    print(body)    results = {"name": body.name, "age": body.age}    return results

• 其它
• 额外信息定义
• 通过 Config schema_extra 的方式实现在文档中展示简单示例

class Item(BaseModel):    name: str    description: Optional[str] = None    price: float    tax: Optional[float] = Noneclass Config: schema_extra = { # 这里就会展示在接口文档中"example": { "name": "lc", "description": "这是描述信息", "price": 300, "tax": 0.7 } }@app.post("/items") def return_item(item: Item): result = {"item": item} return result

表单数据相关 → Form

前端在使用Form表单形式传递数据的时候，其编码格式是采用“特殊”编码，通常要求提交请求头字段Content-Type的方式是application/x-www-form-urlencoded，因此需要fastapi进行特殊处理；具体步骤如下

• 安装表单解析库 → python-multipart
• form数据定义，和Body的定义类似

from fastapi import FastAPI, File, UploadFile, Formfrom typing import List@app.post("/files") async def create_file( file: bytes = File(...), one: List[UploadFile] = File(...), token: str = Form(...) ): return { "filesize": len(file), "token": token, "oen_content_type": [file.content_type for file in one] }

@app.post("/demo/login/")async def login(username: str = Form(...,title="用户名",description="用户名字段描述", max_length=5),                 password: str = Form(...,title="用户密码",description="用户密码字段描述", max_length=20)):    return {"username": username, "password": password}# 多文件的上传@app.post("/sync_file2",summary='File 列表形式的-多文件上传') def sync_file2(files: List[bytes] = File(...)): ''' 基于使用 File 类 运行多文件上传处理 :param files: :return: ''' return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles",summary='UploadFile 形式的-单文件上传') async def uploadfiles(file: UploadFile = File(...)): result = { "filename": file.filename, "content-type": file.content_type, } content = await file.read() with open(f"./{file.filename}", 'wb') as f: f.write(content) return result

请求头相关 → Header()

总体和Body()、Form()的类似，都是在视图函数中进行定义的

@app.get("/demo/header/")# convert_underscores=True 转换为驼峰命名 浏览器一般都是通过驼峰的方式进行数据传递的async def read_items(user_agent: Optional[str] = Header(None,convert_underscores=True), accept_encoding: Optional[str] = Header(None,convert_underscores=True), accept: Optional[str] = Header(None), accept_token: Optional[str] = Header(...,convert_underscores=False), ): return { "user_agent": user_agent, "accept_encoding": accept_encoding, "accept": accept, "token": accept_token,}@app.get("/headerlist/") async def read_headerlist(x_token: List[str] = Header(None)): return {"X-Token values": x_token} # 重名请求头参数

Cookie相关

@app.get("/set_cookie/")def setcookie(response: Response):    response.set_cookie(key="Lbxin", value="Lbxin-1200")    return 'set_cookie ok!'@app.get("/get_cookie") async def Cookier_handel(Lbxin: Optional[str] = Cookie(None)): return { 'Lbxin':Lbxin }

响应相关

响应模型 → response_model

• 基本配置参数
• response_model 是装饰器方法（get，post等的）一个参数，不像之前的所有的参数和请求体，它不属于路径操作参数。
• 可以通过 response_model_exclude 、 response_model_include 等参数对响应模型进行额外的配置。
• response_model_include和response_model_exclude同时使用时会优先处理response_model_include，而response_model_include是存定义的 response_model=UserOut 中进行过滤的，因此会出现输出内容不符合预期的情况
• 避免同时使用 response_model_exclude 和 response_model_include ，除非你清楚它们的优先级和处理逻辑。
• 要保证 response_model_include 里指定的字段在响应模型中是存在的。
• 作用
• 数据过滤：可以制定只返回模型中的部分字段，隐藏敏感或不必要的字段
• 数据验证：避免返回不符合预期的数据
• 自动生成文档

# 定义商品模型class Item(BaseModel): name: str price: float is_offer: bool = None# 定义响应模型，返回商品列表class ItemListResponse(BaseModel): items: List[Item]# 定义API端点，返回商品列表@app.get("/items/", response_model=ItemListResponse) def get_items(): items = [ Item(name="Item 1", price=10.0, is_offer=True), Item(name="Item 2", price=20.0, is_offer=False) ] return {"items": items}# 定义响应模型class ItemResponse(BaseModel): name: str price: float# 定义API端点@app.post("/items/", response_model=ItemResponse) def create_item(item: Item): # 这里可以进行一些业务逻辑处理return item

响应状态码 → status_code

在 fastapi 中，你不用去记住每个状态码的含义，因为利用 fastapi 的内置的

@app.post("/user", response_model=UserOut, status_code=status.HTTP_204_NO_CONTENT)def create_user(user: UserIn):    print(user)    return user

定制响应相关信息(如xml、header、Cookie等)

@app.get("/legacy_with_header_cookie")def legacy_with_header_cookie():    headers = {"X-Xtoken": "LC-1", "Content-Language": "en-US"}    data = """<!--?xml version="1.0"?-->
Apply shampoo here. 
You'll have to use soap here. HERE SOMETHING HEADER YOU DEFINED AND COOKIE """ response = Response(content=data, media_type="application/xml", headers=headers) response.set_cookie(key="cookie_key_lc", value="mrli") return response

更新相关

在创建请求体，更新数据时一般使用PUT请求或者PATCH请求，此外也会结合Pydantic模型的.dict()中使用 exclude_unset 参数实现部分数据更新

• 更新数据的通用步骤
• 获取存储的数据
• 将数据放在Pydantic模型中
• 更新用户设置的值，存储数据中的其它值不改动
• 对原始数据创建副本，通过update参数实现架构接受的数据更新到原始数据中
• 将数据副本转换为可存入数据库的形式
• 数据库数据更新

class Item(BaseModel):    name: Optional[str] = None    description: Optional[str] = None    price: Optional[float] = None    tax: float = 10.5    tags: List[str] = []items = {    "one": {        "name": "apple", "price": 50.3    }}@app.put("/items3", response_model=Item)def upadte_item(name: str, item: Item):    # 检查物品是否存在if name not in items: raise HTTPException(status_code=404, detail="Item not found") stored_item_data = items[name] stored_item_model = Item(**stored_item_data) # 解包为 stored_item_model 对象# 更新部分数据时，可以在 Pydantic 模型的 .dict() 中使用 exclude_unset 参数。upadte_data = item.dict(exclude_unset=True) # 获取只包含 item 对象中的属性 未设置的不会包括# 创建一个新的update_item对象，这个对象是stored_item_model的副本，但是更新了upadte_data中的属性。update_item = stored_item_model.copy(update=upadte_data)# 将update_item对象转换为json格式，然后存储到items中items[name] = jsonable_encoder(update_item) print(items) return update_item

APIRoute实现routes管理

基本上，每个子应用程序都需要包含所有的基本组件，如路由器、中间件异常处理程序以及构建REST API服务所需的所有包，其与常规的应用程序的区别是其上下文路径或URL由处理他们的顶级应用程序定义和决定的

其主要是为了解决fastapi在通过app实例注册路由时的复杂性问题，不能进行统一管理；

一般路由管理是通过APIRouter实例化router后进行统一的router.[methods]进行注册路由，和fastapi实例app注册类似；然后通过app.include_router(XXX.router) 其实就是将所有子Router中的路由都拆解出来并添加到根router后完成路由注册 实现，也可直接在实例化fastapi时进行统一注册管理

• API端点路由注册方式分类
• 基于app实例对象提供的装饰器或函数进行注册
• 基于fastapi提供的 APIRouter 类的实例对象提供的装饰器或函数进行注册
• 本质上是向路由中添加子路由，即理由分组(蓝图模式)
• 通过直接实例化 apiRoute 对象且添加的方式进行注册

APIRouter 内部参数解析

class APIRouter(routing.Router):    def __init__(        self,        *,        prefix: str = "",  # ✅ 表示当前路由分组的url前缀tags: Optional[List[Union[str, Enum]]] = None, # ✅ 表示当前路由分组在可交互文档中所属的分组标签列表。一个 api 端点路由可以属于多个分组dependencies: Optional[Sequence[params.Depends]] = None, # 表示当前路由分组下的依赖项列表。需要注意，这里依赖项列表的返回值不会传递到视图函数内部，也就是说，依赖项的返回值是不会被接收处理的。default_response_class: Type[Response] = Default(JSONResponse), # 表示设置默认响应报文类，默认返回的 JSONResponseresponses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None, # ✅ 表示根据响应体设置不同的响应报文 model 模型callbacks: Optional[List[BaseRoute]] = None, # ✅ 回调函数routes: Optional[List[routing.BaseRoute]] = None, # ✅ 路由组redirect_slashes: bool = True, # 表示是否对路由分组中的斜杠处理进行重定向default: Optional[ASGIApp] = None, dependency_overrides_provider: Optional[Any] = None, # 表示当前的依赖注入提供者，默认指向当前的 app 对象route_class: Type[APIRoute] = APIRoute, # 表示当前 自定义的 APIRoute 类on_startup: Optional[Sequence[Callable[[], Any]]] = None, # 对应 app 中所提供的启动和关闭事件回调函数on_shutdown: Optional[Sequence[Callable[[], Any]]] = None, deprecated: Optional[bool] = None, # ✅ 表示是否标记 API 废弃include_in_schema: bool = True, # ✅ 表示当前路由分组是否显示在可视化交互文档 api 中) -> None:

常规方式

• 实例化定向类，如userRouter类

# -*- coding: utf-8 -*-# /apis/routes/User.pyfrom typing import List from fastapi import APIRouter, HTTPException, Depends from sqlalchemy.orm import Sessionfrom curd import user_curd from schemas import user_schema from utils.db_connect import get_dbuserRouter = APIRouter(tags=['用户相关'])@userRouter.post("/register/", response_model=user_schema.User, summary="用户注册") def create_user(user: user_schema.UserCreate, db: Session = Depends(get_db)): """ 创建用户 """ db_user = user_curd.get_user_by_email(db, email=user.email) if db_user: raise HTTPException(status_code=400, detail="Email already registered") return user_curd.create_user(db=db, user=user)@userRouter.get("/users/", response_model=List[user_schema.User], summary="获取全部的用户列表") def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): """ 获取全部用户 """ users = user_curd.get_users(db, skip=skip, limit=limit) return users@userRouter.get("/users/{user_id}", response_model=user_schema.User, summary="根据用户 ID 来获取用户") def read_user(user_id: int, db: Session = Depends(get_db)): """ 详情页用到 """ db_user = user_curd.get_user(db, user_id=user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found!") return db_user@userRouter.get("/login/", summary="用户登录") async def user_login(email: str, password: str, db: Session = Depends(get_db)): """ 用户登录 API """ db_user = user_curd.get_user_by_email(db, email) hash_password = password +if db_user is None: raise HTTPException(status_code=404, detail="User not found!") if hash_password == db_user.hashed_password: return db_user else: return {"code": 501, "message": "密码错误！"}# 等等userRouter相关...

• 将APIRouter 添加到FastAPI实例

# /apis/routes/User.py# /main.pyfrom fastapi import FastAPI from apis/routes import User,Bolgapp = FastAPI()# 设置一个首页@app.get('/') async def welcome() -> dict: return {"message": "Welcome to my Page"}# 添加FastAPI的API路由app.include_router(User.userRouter) app.include_router(Bolg.blogRouter)

暴力点的方式

from fastapi import FastAPI, Requestfrom fastapi.response import JSONResponsefrom fastapi.routing import APIRouteasync def fastapi_index(): return JSONResponse({'index':'fastapi_index'})async def fastapi_about(): return JSONResponse({'about':'fastapi_about'}) async def info(): return JSONResponse({ "Hello": {"id":123} })app.add_api_route(path='/info', endpoint=info, methods=['GET'])routes = [ # endpoint 指定端点，methods 指定允许访问的方法APIRoute(path='/fastapi/index', endpoint=fastapi_index, methods=['GET','POST']), APIRoute(path='/fastapi/about', endpoint=fastapi_about, methods=['POST']) ]app = FastAPI(routes=routes) app

• 注意点
• 在进行路由URL匹配时，有可能对URL的绑定有一定的要求，如多个地址绑定同一个视图函数，URL同名后的优先级匹配问题等

多应用挂载管理

fastapi项目间应用挂载

当项目庞大时，除了上述的通过APIRouter进行模块划分外，还可以使用主应用挂载子应用的方式进行划分，其具体步骤是：创建主应用及其路由信息 → 创建子应用实例和其路由信息 → 通过 app.mount(path='subapp',app=subapp,name='subapp') 的方式进行主应用挂载子应用关联，其子应用的doc和API相关的都是根据挂载时的path进行交互的

WSGI应用程序间挂载

WSGI应用程序是指如Flask或Django等应用，此时可以通过 WSGI Middleware 的中间件实现fastapi应用无缝挂载 WSGI应用程序

中间件

中间件的执行顺序与添加的顺序一致。即先添加的中间件会先处理请求，后处理响应。

中间件（Middleware）是一种在请求处理前后执行特定代码的机制。它可以用于处理跨域请求、日志记录、请求验证、性能监控等多个方面。

中间件是一个函数或类，它接收一个请求，对其进行处理，然后将请求传递给下一个中间件或路由处理函数。处理完响应后，中间件还可以对响应进行修改。FastAPI 支持使用自定义中间件和内置中间件。

自定义中间件和内置中间件

• 内置中间件

from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()# 允许的源列表origins = [ "http://localhost", "http://localhost:8080", ]# 添加 CORS 中间件app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )@app.get("/") async def read_root(): return {"Hello": "World"}

• 自定义中间件 → 函数式中间件

# 通过装饰器@app.middleware("http")定义，意味着它会被应用到所有的HTTP请求上@app.middleware("http")# 函数会在每个请求到达服务器时被调用async def add_process_time_header(request: Request, call_next): # 在请求处理前执行的代码start_time = time.time() # call_next 调用下一个中间件或路由处理函数response = await call_next(request) # 在请求处理后执行的代码process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) response.headers["mrli"] = "mrli" return response @app.get("/test_middleware") def test_middleware(): return "this is test middleware"

• 自定义中间件 → 类式中间件

类式中间件是一个实现了 __call__ 方法的类。它的优点是可以保存状态，方便在多个请求之间共享数据。

class CustomMiddleware(BaseHTTPMiddleware):    def __init__(self, app, some_config=None):        """        类的构造函数，用于初始化中间件。        :param app: FastAPI 应用实例，中间件需要与应用进行关联。        :param some_config: 可选的配置参数，可以根据需求传入不同的配置信息。        """        super().__init__(app)        self.some_config = some_config        # 这里可以初始化一些状态信息，例如计数器、缓存等self.request_count = 0 # 当有新的请求到达时，FastAPI 会调用 `CustomMiddleware` 实例的 `dispatch` 方法async def dispatch(self, request: Request, call_next): """ 核心的调度方法，处理每个请求。:param request: 接收到的请求对象，包含了请求的各种信息，如 URL、请求方法、请求头、请求体等。:param call_next: 一个异步函数，用于调用下一个中间件或路由处理函数。:return: 处理后的响应对象。""" # 在请求处理前执行的代码# 记录请求开始时间start_time = time.time() # 记录请求数量self.request_count += 1 print(f"当前是第 {self.request_count} 个请求") print(f"请求路径: {request.url.path}，请求方法: {request.method}")try: # 调用下一个中间件或路由处理函数，并等待其返回响应response = await call_next(request) except Exception as e: # 处理请求过程中可能出现的异常import traceback print(f"请求处理过程中出现异常: {e}") traceback.print_exc() # 可以根据异常类型返回不同的错误响应from fastapi.responses import JSONResponse response = JSONResponse(content={"error": "请求处理出错"}, status_code=500)# 在请求处理后执行的代码# 计算请求处理时间process_time = time.time() - start_time # 在响应头中添加处理时间的信息response.headers["X-Process-Time"] = str(process_time) print(f"请求处理时间: {process_time} 秒")return response# 添加中间件到 FastAPI 应用中app.add_middleware(CustomMiddleware, some_config={"key": "value"})@app.get("/") async def read_root(): return {"Hello": "World"}

应用配置信息读取

主要是指应用程序在启动前需要读取相关的配置参数或服务项，而这些配置参数一般需要写入到外部文件或者环境变量中，甚至写入线上的配置中心(如nacos、etcd等)进行统一管理

基于Pydantic和.env环境变量读取参数

通过环境变量的方式读取配置参数是fastapi官方推荐的方式，通过Pydantic可以直接解析出对应的配置参数项，同时提供的参数类型校验等功能(如使用Pydantic中的 Field 在Pydantic模型内部声明校验和定义元数据)；

• 步骤
• 安装读取变量的依赖包
• pip install python-doten
• 定义配置文件.env内容

DEBUG=trueTITLE="FastAPI"DESCRIPTION="FastAPI文档明细描述"vERSION="v1.0.0"

• 定义继承于BaseSettings模型的Settings子类

class Settings(BaseSettings):    debug: bool = False    title: str    description: str    version: strclass Config: env_file = ".env" env_file_encoding = 'utf-8' # 提前进行校验定制@field_validator("version", pre=True) def version_len_check(cls, v: str) -> Optional[str]: if v and len(v) == 0: return None return v

• 定义Settings实例对象，完成.env的解析

settings = Settings()print(settings.debug)print(settings.title)print(settings.description)print(settings.version)

• 也可在实例化过程中制定读取.env文件的方式

settings = Settings(_env_file='.env', _env_file_encoding='utf-8')

• 导入对应的Settings模块并提供给fastapi实例对象使用

settings = Settings(_env_file='.env', _env_file_encoding='utf-8')app = FastAPI(    debug=settings.debug,    title=settings.title,    description=settings.description,    version=settings.version,)

• 可以通过单例模式或 lru_cache 进行读取数据的缓存实现

from pydantic import BaseSettings, field_validatorfrom pydantic.tools import lru_cache@lru_cache()def get_settings():    return Settings()

基于文件读取配置参数

可以通过python自带的configparser来解析读取配置参数，实现配置项参数的自动注入

• 步骤
• 定义配置文件内容，如 conf.int

; conf.int; INI文件是一种简单的配置文件格式，常用于存储程序的设置信息。[fastapi_config]debug = Truetitle = "FastAPI"description = "FastAPI文档明细描述"version = v1.0.0[redis] ip = 127.0.0.1 port = 6379 password = 123456

• 导入configparser模块解析配置文件

#!/usr/bin/evn python# -*- coding: utf-8 -*-from fastapi import FastAPI import configparserconfig = configparser.ConfigParser() config.read('conf.ini', encoding='utf-8')app = FastAPI( debug=bool(config.get('fastapi_config', 'debug')), title=config.get('fastapi_config', 'title'), description=config.get('fastapi_config', 'description'), version=config.get('fastapi_config', 'version'), )if __name__ == "__main__": import uvicorn import osapp_modeel_name = os.path.basename(__file__).replace(".py", "") print(app_modeel_name) uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1', reload=True)

全局异常/错误捕获

HTTPException异常类

集成自Python的Exception类，当在处理请求的过程中遇到错误情况时，可以抛出 HTTPException 来终止当前请求的处理，并返回一个包含特定 HTTP 状态码和错误信息的响应给客户端。

# 定义全局异常处理程序@app.exception_handler(HTTPException) async def http_exception_handler(request, exc): return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail, "custom_message": "An HTTP error occurred"}, headers=exc.headers, )@app.get("/items/{item_id}") async def read_item(item_id: str): if item_id not in ["foo", "bar"]: headers = {"X-Error": "This item was not found"} # 因为是 Python 异常，所以不能 return，只能 raise。raise HTTPException(status_code=404, detail="Item not found", headers=headers) return {"item": item_id}

exception_handlers

exception_handlers 参数主要用于捕获在执行业务逻辑处理过程中抛出的各种异常

from fastapi import FastAPIfrom starlette.responses import JSONResponseasync def exception_not_found(request, exc): return JSONResponse({ "code": exc.status_code, "error": "没有定义这个请求地址"}, status_code=exc.status_code)exception_handlers = { 404: exception_not_found, }app = FastAPI(exception_handlers=exception_handlers)if __name__ == "__main__": import uvicorn import osapp_modeel_name = os.path.basename(__file__).replace(".py", "") print(app_modeel_name) uvicorn.run(f"{app_modeel_name}:app", host='127.0.0.1', reload=True)

高级特性

存储相关

Redis

• 基本概念
• 数据存储形式 ：以键值对的形式存储数据(key-value)，键通常是字符串，值可以是多种数据结构，如string、Hash、List、Set和Sorted Set等
• 内存数据库 ：一般是将数据存储在内存中，因此其读(110000次/s)写(81000次/s)速度非常快，当然为了防止数据丢失，Redis也提供了持久化机制，将内存中的数据保存的磁盘上
• 应用场景
• 缓存：将经常访问的数据存储到Redis，减少对数据库的访问压力和提高系统的响应速度
• 计数器：利用了Redis的原子性操作
• 分布式锁：保证同一时间只有一个客户端可以访问共享数据资源
• 消息队列：利用了Redis的列表数据结构实现简单的消息队列，支持生产者-消费者模式

应用示例

from starlette.requests import Requestfrom fastapi import FastAPI, Queryfrom aioredis import Redis,create_redis_poolapp = FastAPI()# redis连接池# 如果 Redis 服务器需要用户名和密码进行身份验证，你需要在 URL 中包含用户名和密码，# 格式为 redis://username:password@host:port/dbasync def create_redis() -> Redis: # redis://[username]:[password@]localhost:port/db# redis:// 是 Redis URL 的协议前缀，表示这是一个 Redis URL。# /db 是数据库编号，1 表示你想要连接到 Redis 服务器的第一个数据库（Redis 支持多个数据库，编号从 0 开始）return await create_redis_pool(f"redis://127.0.0.1:6379/1?encoding=utf-8")@app.get('/test',summary='test redis') async def test_redis(request: Request,num: int = Query(123,title='参数 num')): # 将数据存入 redisawait request.app.state.redis.set("test", num)# 从redis中获取数据value = await request.app.state.redis.get("test") print(value,222,type(value)) return {"value": value}@app.on_event('startup') async def setup_event(): app.state.redis = await create_redis() print('init redis success')@app.on_event('shutdown') async def shutdown_event(): app.state.redis.close() await app.state.redis.wait_closed() print('close redis success')

依赖注入

在 FastAPI 中，依赖注入（Dependency Injection）是一个核心特性，它允许你将代码拆分成更小、更易测试和复用的部分。依赖注入的核心思想是将一个对象所依赖的其他对象以参数的形式传递进来，而不是在对象内部直接创建。

Depends 是 FastAPI 提供的一个用于声明依赖项的工具函数。是为了让Fastapi可以正确识别和处理依赖项，实现自动参数注入和实例创建；

当你使用 Depends(SomeClass) 或者 Depends(some_function) 时，FastAPI 会在处理请求时自动调用 SomeClass 的构造函数（如果是类）或者 some_function （如果是函数），并将所需的参数传递进去，然后把返回值注入到路径操作函数中

依赖项可以是一个普通函数，也可以是一个类的方法，依赖项函数通常接受一些参数，这些参数可以是路径参数、查询参数、请求体等

• 适用场景
• 共享业务逻辑(代码逻辑复用)
• 共享数据库连接
• 实现安全、验证、角色权限等

全局依赖项注入

def verify_token(token: str = Header(...)):    if token != "mrli_token":        raise HTTPException(status_code=400, detail="token is invalid")app = FastAPI(dependencies=[Depends(verify_token)])@app.get("/items") def read_items(): return fake_db_items@app.get("/users") def read_users(): return "this is test return user"

公共逻辑注入

fake_db_items = [{"city": "beijing"}, {"city": "shanghai"}, {"city": "guangzhou"}]def verify_token(token: str = Header(...)): if token != "mrli_token": raise HTTPException(status_code=400, detail="token is invalid")def verify_key(key: str = Header(...)): if key != "mrli_key": raise HTTPException(status_code=400, detail="key is invalid") return key# 必在声明路径操作函数的参数时使用 Depends，而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list@app.get("/items", dependencies=[Depends(verify_token), Depends(verify_key)]) def read_items(): return fake_db_itemsif __name__ == '__main__': import uvicorn uvicorn.run("main:app", reload=True, debug=True)

数据依赖项的注入

# 定义一个依赖项函数async def common_parameters(q: str = None, skip: int = 0, limit: int = 100): return {"q": q, "skip": skip, "limit": limit}# `dict` 是一种内置的数据类型，全称为字典（dictionary），它是一种无序、可变且可迭代的数据结构，用于存储键 - 值对（key - value pairs）。# 依赖注入中，`dict` 作为类型注解用于指定依赖项函数返回的数据类型是字典，帮助开发者和 FastAPI 明确数据的结构和类型，提高代码的可读性和可维护性# 这里 `commons` 参数的类型注解 `dict` 表示它期望接收一个字典类型的数据，该数据由 `common_parameters` 函数返回@app.get("/items/") async def read_items(commons: dict = Depends(common_parameters)): return commons# 定义一个依赖项类# 类的 `__init__` 方法的参数来从请求中提取相应的参数（如查询参数、路径参数等），并创建 `CommonQueryParams` 类的一个实例，将其作为依赖项注入到路径操作函数中。class CommonQueryParams: def __init__(self, q: str = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit@app.get("/items/class/") async def read_items(commons: CommonQueryParams = Depends()): return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}@app.get("/items3/{item_id}", response_model=Item) def read_item(item_id: str): return items[item_id]

异步编程

fastapi 框架最大的特性就是异步支持，且同时支持 ASGI 协议和 WSGI 的应用

计算机的任务主要有两类，分别是 计算密集型任务 和 IO密集型任务(如输入/输出阻塞、磁盘IO、网络请求IO) ；而程序处理并发任务问题的常见方案是多线程和多进程，引入多线程和多进程方式在某种程度上可以实现多任务并发执行，线程间相互独立执行互不影响，因此 对于IO型任务通常通过多线程调度实现并发、对于计算密集型任务则使用多进程来实现并发

ASGI协议简介

是为了规范支持异步的Python Web服务器、框架和应用之间的通信而定制的，同时囊括了 同步和异步 应用的通信规范，并且 向后兼容 WSGI协议。由于最新的HTTP支持异步长连接，而传统的WSGI应用支持单次同步调用，即仅在接收一个请求后返回响应，从而无法支持HTTP长轮询或WebSocket连接。在Python 3.5增加async/await特性之后，基于asyncio异步协程的应用编程变得更加方便。ASGI协议规范就是用于asyncio框架中底层服务器/应用程序的接口。

异步IO

系统遇到IO任务时，CPU不会等待IO任务执行完成，而是直接继续后续任务执行，其本质是 基于事件触发机制来实现异步回调 ，在IO处理上主要采用IO服用机制来实现非阻塞操作；

这种异步非阻塞是在一个单线程内完成的，在一个线程内可以 高效处理更多的IO任务 ，这就是异步IO的魅力所在

推荐文献

• fastapi文档
• Fastapi框架学习笔记
• FastAPI 学习之路
• code