[toc]

FastAPI笔记2

Pydantic 模型

Pydantic 是一个用于数据验证和序列化的 Python 数据模型库。

它在 FastAPI 中广泛使用，用于定义请求体、响应体和其他数据模型，提供了强大的类型检查和自动文档生成功能。

定义 Pydantic 模型

使用 Pydantic 定义一个模型非常简单，只需创建一个继承自 pydantic.BaseModel 的类，并在其中定义字段。字段的类型可以是任何有效的 Python 类型，也可以是 Pydantic 内置的类型。

# 导入Pydantic包的 BaseModel类
from pydantic import BaseModel
# 定义Item模型类及其属性字段，继承BaseModel类
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

以上代码中，定义了一个名为 Item 的 Pydantic 模型类。其中name 和 price 是必需的字段，而 description 和 tax 是可选的字段，其默认值为 None。

使用 Pydantic 模型

请求体参数验证和请求参数转换

在 FastAPI 中，可以将 Pydantic 模型用作请求体（Request Body），以自动验证和解析客户端发送的数据。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

上面代码中，函数接受一个名为 item 的参数，其类型是 Item 模型。FastAPI 将自动验证传入的 JSON 数据是否符合 Item 模型的定义。若符合则将JSON 数据自动转换为 Item 类型的实例对象。

参数验证

Pydantic 模型还可以用于验证查询参数、路径参数等。

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.get("/items/")
def read_item(item: Item, q: str = Query(..., max_length=10)):
    return {"item": item, "q": q}

通过使用 Query 函数，我们还可以为参数指定更多的验证规则，如最大长度限制。

返回 Pydantic 模型对象

在 FastAPI 中，你可以直接返回 Pydantic 模型对象作为路由操作函数的返回值。

FastAPI 会自动将模型对象转换为 JSON 格式的数据并响应给客户端。

from fastapi import APIRouter
from pydantic import BaseModel
router = APIRouter()

# 定义 Pydantic 数据模型类。该类需要继承pydantic包中的BaseModel类。
# 并且类中的属性需要符合Pydantic数据模型的要求。
class Item(BaseModel):
    name: str
    description: str = None

@router.post("/items/")
def get_item(item: Item):
    return item  # 返回数据模型类数据

返回的数据格式如下

json

{
  "name": "xxxx",
  "description": "xxxxxxx",
}

上面代码中，路由操作函数 get_item 返回一个 Item 对象。
FastAPI 会自动将该 Item 对象转换为 JSON 格式的响应体，包含 name、description 字段。

序列化反序列化模型对象

序列化：将模型对象转换为 JSON 格式的字符串。
反序列化：将 JSON 格式的字符串转换为模型对象。

model_dump() - 模型转字典

将 Pydantic 模型实例转换为 Python 字典

from pydantic import BaseModel
class UserDTO(BaseModel):
    id: int = None
    name: str = None

# 转换为字典
user_dto_dict = UserDTO.model_dump()
# 输出: {"id": 1, "name": "111"}

model_dump_json() - 模型转 JSON

from pydantic import BaseModel
class UserDTO(BaseModel):
    id: int = None
    name: str = None

# 转换为 JSON 字符串
user_dto_json = UserDTO.model_dump_json()
# 输出: {"id": 1, "name": "111"}

序列化中排除/包含特定字段

# 排除 id 字段
aaa = UserDTO.model_dump(exclude={"id"})

# 只包含 name 字段
aaa = UserDTO.model_dump(include={"name"})

序列化别名处理

from pydantic import BaseModel, Field
# 如果定义了别名，在序列化时会使用别名作为键名
class UserDTO(BaseModel):
    id: int = None
    name: str = None
    # 使用 Field 类为字段指定别名
    age: int = Field(..., alias="userAge")

# 默认序列化
aaa = UserDTO.model_dump(by_alias=False)  # {...,"age": 10} 
# 使用别名进行序列化
aaa = UserDTO.model_dump(by_alias=True)  # {...,"userAge": 10}

自定义序列化方法

from pydantic import BaseModel, Field
# 如果定义了别名，在序列化时会使用别名作为键名
class UserDTO(BaseModel):
    id: int = None
    name: str = None
    
    @field_serializer("id")
    def serialize_id(self, id: int) -> str:
        # 将 id 转换为字符串格式
        return f"USER_{id}"
    
    @field_serializer("name")
    def serialize_name(self, name: str) -> str:
        # 将 name 转换为其他格式
        return f"NAME_{name}"

aaa = UserDTO.model_dump()
# 输出: {"id": "USER_1", "name": "NAME_xiaoming"}

复杂嵌套模型的序列化

python

from pydantic import BaseModel
from typing import List, Optional
# 问题模型(里面嵌套了选项模型)
class QuestionDTO(BaseModel):
    id: int
    content: str
    options: List[OptionDTO]

# 选项模型
class OptionDTO(BaseModel):
    id: int
    content: str

# 创建嵌套模型实例
question = QuestionDTO(
    id=1,
    content="2+2=?",
    options=[
        OptionDTO(id=1, content="3"),
        OptionDTO(id=2, content="4")
    ]
)

# 序列化嵌套模型
question_dict = question.model_dump(by_alias=True)
# 输出: {"id": 1, "content": "2+2=?", "options": [{"id": 1, "content": "3"}, {"id": 2, "content": "4"}]}

Pydantic 模型和 SqlAlchemy 模型的转换

model_validate() 方法可以将 SqlAlchemy 模型实例转换为 Pydantic 模型实例。

如果需要，model_validate() 方法可以设置 exclude 参数排除敏感字段。

python

from modules.module_exam.model.mp_user_model import MpUserModel
from modules.module_exam.dto.mp_user_dto import MpUserDTO
# 假设定义了一个Pydantic模型类MpUserDTO，用于表示用户数据
class MpUserDTO(BaseModel):
    id: int = None
    name: str = None
    age: int = None

# 假设从数据库获取了一个SQLAlchemy模型实例
db_user = MpUserModel(
    id=1,
    name="张三",
    age=25,
)

# 方法1: 直接使用model_validate方法
pydantic_user = MpUserDTO.model_validate(db_user)

# 方法2: 如果需要，可以传递exclude参数排除敏感字段
pydantic_user_safe = MpUserDTO.model_validate(db_user, exclude={"id"})

FastAPI 路径操作依赖项

FastAPI 提供了路径操作依赖项的机制，允许你在路由处理函数执行之前或之后运行一些额外的逻辑。

依赖项

依赖项是在路由操作函数执行前或后运行的可复用的函数或对象。它们被用于执行一些通用的逻辑，如验证、身份验证、数据库连接等。

在 FastAPI 中，依赖项通常用于两个方面：

预处理（Before）依赖项：在路由操作函数执行前运行，用于预处理输入数据，验证请求等。
后处理（After）依赖项：在路由操作函数执行后运行，用于执行一些后处理逻辑，如日志记录、清理等。

预处理（Before）依赖项

from fastapi import Depends, FastAPI

app = FastAPI()

# 依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 路由操作函数
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

上面代码中依赖项函数接受参数 q、skip 和 limit，并返回一个包含这些参数的字典数据。
commons: dict = Depends(common_parameters)路由操作函数接收依赖项函数的返回值作为参数。

后处理（After）依赖项

from fastapi import Depends, FastAPI
app = FastAPI()

# 后处理依赖项函数
async def after_request():
    # 这里可以执行一些后处理逻辑，比如记录日志
    pass

# 路由操作函数
@app.get("/items/", response_model=dict)
async def read_items_after(request: dict = Depends(after_request)):
    return {"message": "Items returned successfully"}

多个依赖项函数组合使用

from fastapi import Depends, FastAPI
app = FastAPI()

# 依赖项函数1
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 依赖项函数2, 依赖于依赖项函数common_parameters
def verify_token(token: str = Depends(common_parameters)):
    return token

# 路由操作函数, 依赖于依赖项函数verify_token
@app.get("/items/")
async def read_items(token: dict = Depends(verify_token)):
    return token

异步依赖项

依赖项函数和路由处理函数可以是异步的，允许在它们内部执行异步操作。

from fastapi import Depends, FastAPI
from typing import Optional
import asyncio

app = FastAPI()

# 异步依赖项函数
async def get_token():
    # 模拟异步操作
    await asyncio.sleep(2)
    return "fake-token"

# 异步路由操作函数
@app.get("/items/")
async def read_items(token: Optional[str] = Depends(get_token)):
    return {"token": token}

上面代码中，get_token 是一个异步的依赖项函数，模拟了一个异步操作。

中间件

在 FastAPI 中，中间件是一个在每个请求被处理之前以及每个响应返回之前运行的函数。它可以接收应用程序的每一个请求，执行一些操作，然后将请求传递给应用程序的其他部分，最后获取并返回响应

中间件通常用于执行一些通用的逻辑，如日志记录、身份验证、性能监控等。

中间件的执行流程如下

客户端发送请求到 FastAPI 应用。
FastAPI 应用将请求传递给第一个定义的中间件函数。
中间件函数可以在请求处理前执行一些逻辑，比如记录请求时间。
然后中间件函数将请求传递给下一个中间件函数或路由处理函数。
如果有更多的中间件函数定义，重复步骤 3-4，直到最后一个中间件函数调用路由处理函数。
路由处理函数执行，生成响应对象。
每个中间件函数在返回响应之前可以执行一些逻辑，比如添加响应头信息。
最后，FastAPI 应用将响应结果返回给客户端。

FastAPI 提供了多种实现中间件的方式，以下是常用的几种。

函数式中间件（推荐用于简单场景）

函数式中间件是一种基于函数的中间件实现方式，它允许你定义一个普通的异步函数，用于处理每个请求。函数中间件需要接收两个参数：request 和 call_next。

request：是当前的请求对象，包含了请求的信息和数据。
call_next：一个异步函数，这个函数将请求传递给相应的路径操作函数，然后返回由该路径操作函数生成的响应对象。你可以在返回响应对象之前进一步修改它。

函数式中间件有两种写法：装饰器写法和自定义函数写法。

装饰器写法

将函数和装饰器 @app.middleware("http") 结合起来，就可以将函数快速注册为中间件。

示例如下

python

from fastapi import FastAPI, Request

app = FastAPI()

# 定义中间件函数
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    # 在请求处理前执行一些逻辑，比如记录请求时间
    start_time = time.time()
    
    # 调用下一个中间件或路由处理函数
    response = await call_next(request)
    
    # 在请求处理后执行一些逻辑，比如记录响应时间
    process_time = time.time() - start_time
    # 将处理时间添加到响应对象中
    response.headers["X-Process-Time"] = str(process_time)
    
    # 返回响应对象
    return response

自定义函数写法

先自定义一个函数，然后再将这个函数注册为中间件。

logger_middleware.py 文件代码如下

python

from fastapi import Request
import time
from loguru import logger
from starlette.responses import Response
from typing import Callable, Any

async def logger_middleware(request: Request, call_next: Callable) -> Response:
    # 请求开始时间
    start_time = time.time()

    # 记录请求信息
    logger.info(f"Request: {request.method} {request.url.path}")

    # 继续处理请求，并获取响应对象
    response = await call_next(request)

    # 计算处理时间
    process_time = time.time() - start_time

    # 记录响应信息
    logger.info(f"Response: {response.status_code} - {process_time:.4f}s")

    # 添加响应时间到头部
    response.headers["X-Process-Time"] = str(process_time)
    return response

main.py 文件中注册中间件。代码如下

python

from fastapi import FastAPI
app = FastAPI()

# 导入函数式中间件
from middlewares.logger_middleware import logger_middleware
# 使用app.middleware("http")装饰器注册函数式中间件
app.middleware("http")(logger_middleware)

类中间件（推荐用于复杂场景）

类中间件是一种基于类的中间件实现方式，它允许你将中间件逻辑封装在一个类中，提供更灵活的配置和管理。

类中间件需要继承自 BaseHTTPMiddleware 类，并重写 dispatch 方法。在 dispatch 方法中，你可以实现自定义的中间件逻辑，比如请求预处理、响应后处理等。
然后通过 app.add_middleware(xxxx) 方法注册类中间件。

logger_middleware.py 文件代码示例如下

python

from fastapi import Request
import time
from loguru import logger
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
from typing import Callable, Any

class LoggerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Callable) -> Response:
        # 请求开始时间
        start_time = time.time()
        # 记录请求信息
        logger.info(f"Request: {request.method} {request.url.path}")
        # 继续处理请求，并获取响应对象
        response = await call_next(request)

        # 计算处理时间
        process_time = time.time() - start_time
        # 记录响应信息
        logger.info(f"Response: {response.status_code} - {process_time:.4f}s")
        # 添加响应时间到头部
        response.headers["X-Process-Time"] = str(process_time)
        return response

main.py 文件中注册中间件。代码如下

python

from fastapi import FastAPI
app = FastAPI()

# 导入类式中间件
from middlewares.logger_middleware import LoggerMiddleware
# 使用app.add_middleware注册类中间件
app.add_middleware(LoggerMiddleware)

常用中间件

CORS 中间件

CORS（跨域资源共享）中间件用于处理跨域请求。

FastAPI中内置了一个CORSMiddleware中间件，用于在HTTP响应中添加必要的CORS头信息，使得浏览器能够允许来自不同源的请求访问你的API。

在main.py文件中注册CORS中间件

python

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()

# 注册CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有HTTP方法
    allow_headers=["*"],  # 允许所有请求头
)

日志中间件

日志中间件用于记录请求和响应的详细信息，帮助开发人员调试和监控应用程序。有以下两种写法。

python

from fastapi import Request
import time
from loguru import logger
from starlette.responses import Response
from typing import Callable

async def LoggerMiddleware(request: Request, call_next: Callable) -> Response:
    # 请求开始时间
    start_time = time.time()
    # 记录请求信息
    logger.info(f"Request: {request.method} {request.url.path}")

    # 继续处理请求，并获取响应对象
    response = await call_next(request)

    # 计算处理时间
    process_time = time.time() - start_time
    # 记录响应信息
    logger.info(f"Response: {response.status_code} - {process_time:.4f}s")
    # 添加响应时间到头部
    response.headers["X-Process-Time"] = str(process_time)
    return response


# 在main.py文件中注册中间件..........
from middlewares.logger_middleware import LoggerMiddleware
app.middleware("http")(LoggerMiddleware)

身份认证中间件

身份认证中间件用于验证请求中的身份信息，确保只有合法用户才能访问受保护的资源。

auth_middleware.py

python

from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.responses import Response
from typing import Callable

# 认证中间件
async def AuthMiddleware(request: Request, call_next: Callable) -> Response:
    print("AuthMiddleware========================")

    # 精确匹配的路径
    exact_paths = ["/", "/login", "/docs", "/redoc", "/openapi.json"]

    # 测试路径前缀（支持通配符效果）
    test_path_prefixes = [
        "/test/",  # 匹配所有以 /test/ 开头的路径
        "/api/test/"  # 匹配所有以 /api/test/ 开头的路径
    ]

    # 检查是否是精确匹配的路径
    if request.url.path in exact_paths:
        return await call_next(request)

    # 检查是否是测试路径前缀
    for prefix in test_path_prefixes:
        if request.url.path.startswith(prefix):
            return await call_next(request)

    # 获取 Authorization 头
    auth_header = request.headers.get("Authorization")

    # 验证 token
    if not auth_header or not auth_header.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"detail": "未授权访问，需要有效的token"}
        )

    # 这里可以添加更复杂的 token 验证逻辑
    token = auth_header.split(" ")[1]
    if token != "valid-token":  # 实际应用中应该验证真实的 token
        return JSONResponse(
            status_code=401,
            content={"detail": "无效的 token"}
        )

    # 验证通过，继续处理请求
    return await call_next(request)

python

# main.py 文件中注册中间件
from middlewares.auth_middleware import AuthMiddleware
app.middleware("http")(AuthMiddleware)

异常处理中间件

异常处理中间件用于捕获应用程序中未处理的异常，并返回统一的错误响应。

在main.py文件中

python

from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse

class ExceptionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            # 尝试处理请求
            response = await call_next(request)
            return response
        except ValueError as e:
            # 处理特定类型的异常
            return JSONResponse(
                status_code=400,
                content={"code": 400, "message": str(e), "data": None}
            )
        except Exception as e:
            # 处理所有其他未捕获的异常
            return JSONResponse(
                status_code=500,
                content={"code": 500, "message": f"服务器内部错误: {str(e)}", "data": None}
            )

app = FastAPI()
app.add_middleware(ExceptionMiddleware)

GZip 压缩中间件

GZipMiddleware 是 FastAPI/Starlette 框架提供的一个内置中间件，用于自动压缩 HTTP 响应内容。它可以大幅减少传输数据的大小，提高 API 性能，特别是在处理大型响应数据时效果显著。

python

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware
app = FastAPI()

# 添加 GZip 中间件，压缩大于 1000 字节的响应
app.add_middleware(GZipMiddleware, minimum_size=1000)

统一返回方式

在 FastAPI 中，统一返回格式的封装是后端服务接口设计的重要部分。主要有以下几种主流的统一返回方式。

方式1 装饰器

使用 Python 装饰器统一处理返回格式

python

from functools import wraps
from fastapi import Response
import json

# 定义统一返回格式的装饰器
def standard_response(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        result = await func(*args, **kwargs)
        return {"code": 200, "data": result, "message": "success"}
    return wrapper

# 示例
@app.get("/items/")
@standard_response
async def get_items():
    return [{"id": 1, "name": "item1"}]  # 直接返回业务数据，装饰器自动包装

方式2 Pydantic 模型类

使用 Pydantic 模型类定义统一的响应类。

FastAPI会自动将Pydantic模型类转换为JSON格式的数据，并响应给客户端。

python

from pydantic import BaseModel
from typing import Generic, TypeVar, Optional, Any

# 定义泛型变量
T = TypeVar('T')

# 定义Pydantic模型类
class ResponseModel(BaseModel, Generic[T]):
    code: int = 200
    message: str = "success"
    data: Optional[T] = None

# 示例
@app.get("/items/", response_model=ResponseModel[List[Item]])
async def get_items():
    items = [{"id": 1, "name": "item1"}]
    return ResponseModel(data=items)

方式3 工具类

创建专门的响应工具类，在接口中显式调用，封装返回格式。

python

from fastapi.responses import JSONResponse

# 定义响应工具类ResponseUtil
class ResponseUtil:
    # 定义成功响应的静态方法
    @staticmethod
    def success(data=None, message="success"):
        return {"code": 200, "data": data, "message": message}
    
    # 定义错误响应的静态方法
    @staticmethod
    def error(code=500, message="error", as_dict=False):
        result = {"code": code, "data": None, "message": message}
        return result if as_dict else JSONResponse(content=result, status_code=code)

# 示例
@app.get("/items/")
async def get_items():
    items = [{"id": 1, "name": "item1"}]
    return ResponseUtil.success(data=items)

方式4 依赖注入

使用 FastAPI 的依赖注入功能提供统一响应格式

python

from fastapi import Depends
# 定义依赖项函数get_response_formatter
def get_response_formatter():
    def format_response(data, message="success"):
        return {"code": 200, "data": data, "message": message}
    return format_response

# 示例
@app.get("/items/")
async def get_items(formatter=Depends(get_response_formatter)):
    items = [{"id": 1, "name": "item1"}]
    return formatter(items)

全局异常处理

在 FastAPI 中，全局异常处理是指在应用程序中统一处理所有未捕获的异常，避免在接口层暴露敏感信息。

以下是主流的全局异常处理方式。

方式1 全局异常处理器

使用 FastAPI 的 @app.exception_handler() 装饰器注册异常处理器函数，处理所有未捕获的异常。

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 处理特定异常
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"code": 400, "message": str(exc), "data": None}
    )

# 处理所有异常
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": f"服务器内部错误: {str(exc)}", "data": None}
    )

上面例子中，一旦程序中发生异常。会被全局异常处理器捕获。根据异常类型，返回不同的响应信息。

方式2 中间件

使用 FastAPI 中间件捕获和处理异常。

在exception_middleware.py文件中定义全局异常处理中间件

python

from fastapi import Request
from fastapi.responses import JSONResponse
from typing import Callable

# 全局异常处理中间件
async def ExceptionMiddleware(request: Request, call_next: Callable):
    print("ExceptionMiddleware===============")
    try:
        # 正常处理请求
        response = await call_next(request)
        return response
    except Exception as e:
        # 捕获所有异常并处理

        # 若异常信息包含code和message属性
        if hasattr(e, 'code') and hasattr(e, 'message'):
            return JSONResponse(
                status_code=e.code,
                content={"code": e.code, "message": e.message}
            )
        else:
            # 将异常信息转换为JSON响应
            return JSONResponse(
                status_code=500,
                content={"code": 500,"message": f"服务异常信息= {str(e)}"}
            )

在main.py文件中注册全局异常中间件

python

from fastapi import FastAPI
from exception_middleware import ExceptionMiddleware

app = FastAPI()
app.middleware("http")(ExceptionMiddleware)

方式3 装饰器

使用装饰器包装路由处理函数

python

from functools import wraps
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# 定义异常处理装饰器
def handle_exceptions(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except ValueError as e:
            return JSONResponse(
                status_code=400,
                content={"code": 400, "message": str(e), "data": None}
            )
        except Exception as e:
            return JSONResponse(
                status_code=500,
                content={"code": 500, "message": f"服务器内部错误: {str(e)}", "data": None}
            )
    return wrapper


# 定义函数，使用装饰器处理异常
@app.get("/items/")
@handle_exceptions
async def get_items():
    # 业务逻辑
    pass

异步编程

在 FastAPI 中，异步编程主要依赖于 Python 的异步编程特性，特别是 async 和 await 关键字。这些特性允许你编写非阻塞的代码，从而提高应用的性能和响应速度。

使用 async 和 await 关键字

当你调用一个支持异步的库时，你应该使用 async def 来定义你的函数，并在需要等待结果的地方使用 await。这样，FastAPI 就能够在等待时执行其他任务，提高并发性能。

如果你使用的库不支持异步操作，或者你的函数只是执行计算并不需要等待 I/O 操作，你可以使用普通的 def 来定义函数。

基本示例

在 FastAPI 中，你可以通过定义异步路径操作函数来实现并发处理。

python

from fastapi import FastAPI
app = FastAPI()

# 定义异步路径操作函数
@app.get('/')
async def read_results():
   # await 等待异步函数some_library 返回结果之后，执行后面代码
   results = await some_library()
   return results

FastAPI笔记2 ​

Pydantic 模型 ​

定义 Pydantic 模型 ​

使用 Pydantic 模型 ​

返回 Pydantic 模型对象 ​

序列化 反序列化 模型对象 ​

model_dump() - 模型转字典 ​

model_dump_json() - 模型转 JSON ​

序列化中排除/包含特定字段 ​

序列化 别名处理 ​

自定义序列化方法 ​

复杂嵌套模型的序列化 ​

Pydantic 模型和 SqlAlchemy 模型的转换 ​

FastAPI 路径操作依赖项 ​

依赖项 ​

预处理（Before）依赖项 ​

后处理（After）依赖项 ​

多个依赖项函数组合使用 ​

异步依赖项 ​

中间件 ​

函数式中间件（推荐用于简单场景） ​

类中间件（推荐用于复杂场景） ​

常用中间件 ​

CORS 中间件 ​

日志中间件 ​

身份认证中间件 ​

异常处理中间件 ​

GZip 压缩中间件 ​

统一返回方式 ​

方式1 装饰器 ​

方式2 Pydantic 模型类 ​

方式3 工具类 ​

方式4 依赖注入 ​

推荐方式 Pydantic 模型类 + 工具类 ​

全局异常处理 ​

方式1 全局异常处理器 ​

方式2 中间件 ​

方式3 装饰器 ​

异步编程 ​

FastAPI笔记2

Pydantic 模型

定义 Pydantic 模型

使用 Pydantic 模型

返回 Pydantic 模型对象

序列化反序列化模型对象

model_dump() - 模型转字典

model_dump_json() - 模型转 JSON

序列化中排除/包含特定字段

序列化别名处理

自定义序列化方法

复杂嵌套模型的序列化

Pydantic 模型和 SqlAlchemy 模型的转换

FastAPI 路径操作依赖项

依赖项

预处理（Before）依赖项

后处理（After）依赖项

多个依赖项函数组合使用

异步依赖项

中间件

函数式中间件（推荐用于简单场景）

类中间件（推荐用于复杂场景）

常用中间件

CORS 中间件

日志中间件

身份认证中间件

异常处理中间件

GZip 压缩中间件

统一返回方式

方式1 装饰器

方式2 Pydantic 模型类

方式3 工具类

方式4 依赖注入

推荐方式 Pydantic 模型类 + 工具类

全局异常处理

方式1 全局异常处理器

方式2 中间件

方式3 装饰器

异步编程