前言

FastAPI 不仅以其高性能和易用性著称，还内置了丰富的高级特性，使得开发者可以快速组织和封装各种功能。在本篇博客中，我们将深入探讨 FastAPI 的一些高级特性，包括错误处理、中间件和应用生命周期管理。通过这些特性，开发者可以在应用的不同阶段和层面上挂载自定义逻辑，从而实现更灵活和强大的应用架构。

• 错误处理：FastAPI 提供了简便的错误处理机制，使得开发者可以轻松地定义和管理自定义错误响应。
• 中间件：中间件可以在请求到达路由之前和响应返回客户端之前执行自定义逻辑，适用于日志记录、鉴权、限流等功能。
• 应用生命周期：FastAPI 允许在应用的启动和关闭过程中执行特定的逻辑，这对于资源初始化和清理非常有用。

通过这些内置的入口和钩子函数，FastAPI 为开发者提供了强大的工具，能够快速组织和封装各种功能，从而打造出高效、可维护的应用程序。接下来，我们将通过具体示例来展示这些高级特性的使用方法和优势。

错误处理

python

复制代码
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @File: err_handle.py
# @Desc: { 错误处理demo }
# @Date: 2024/05/21 11:07
from typing import Optional

import uvicorn
from fastapi import FastAPI, Request, Depends, Query
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field
from starlette.responses import JSONResponse
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from loguru import logger


async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    """全局异常处理"""
    if isinstance(exc, BizException):
        return JSONResponse(
            status_code=200,
            content={"code": exc.code, "msg": exc.msg, "data": {}}
        )
    elif isinstance(exc, RequestValidationError):
        return await request_validation_exception_handler(request, exc)
    else:
        logger.exception("Uncaught exception：sentry 告警 飞书通知")
        return JSONResponse(content={"code": 5000, "msg": "Internal Server Error", "data": {}}, status_code=500)


# 添加错误处理器
app = FastAPI(exception_handlers={Exception: global_exception_handler})
# app.add_exception_handler(Exception, global_exception_handler)


class BizException(Exception):
    def __init__(self, code=0, msg=""):
        self.code = code
        self.msg = msg


# @app.exception_handler(Exception)
# async def exception_handler(request: Request, exc: Exception):
#     return await global_exception_handler(request, exc)


class ErrorModel(BaseModel):
    error_num: int = Field(Query(description="错误编码"))
    error_msg: Optional[str] = Field(Query(description="错误信息"))


@app.get("/error_handle")
async def error_handle(req_model: ErrorModel = Depends(ErrorModel)):
    a = 1 / 0
    return req_model.model_dump()


@app.get("/biz_exception")
async def biz_exception():
    order_num = 0
    if order_num <= 0:
        raise BizException(msg="订单不足")

    order_num -= 1
    return order_num


def main():
    uvicorn.run(app)


if __name__ == '__main__':
    main()

这里通过 @app.exception_handler(Exception) 装饰器方式进行异常捕获，这里可以指定特定的异常，我这里是指定 Exception 统一异常处理，在异常处理函数里面具体判断异常类型进行不同处理。这样业务代码中就可以通过抛异常来结束业务逻辑，一些未知未处理的异常也可以统一进行代码告警，集中在一起非常方便的处理。

中间件

装饰器方式

python

复制代码
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @File: middleware_demo.py
# @Desc: { 模块描述 }
# @Date: 2024/05/21 11:07
import time
import uvicorn
from fastapi import FastAPI, Request

app = FastAPI(description="中间件demo")


@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


@app.get("/middleware_demo")
async def middleware_demo():
    return {"demo": "middleware_demo"}


def main():
    uvicorn.run(app)


if __name__ == '__main__':
    main()

添加中间件函数方式

python

复制代码
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware

class APIProcessTimeMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, 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


app = FastAPI(description="中间件demo")
app.add_middleware(APIProcessTimeMiddleware)

参数传递方式

python

复制代码
class APIProcessTimeMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, 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


# app = FastAPI(description="中间件demo")
app = FastAPI(description="中间件demo", middleware=[Middleware(APIProcessTimeMiddleware)])
# app.add_middleware(APIProcessTimeMiddleware)

参数传递方式要使用 Middlware 包装一层。

中间件可以在请求前后处理一些事情，例如

• 日志记录：记录每个请求的详细信息，包括请求时间、请求路径、请求参数、响应状态等。

• 压缩/解压缩：对请求和响应进行压缩处理，以减少数据传输量。

• 会话管理：管理用户会话信息，维护用户登录状态。

• 数据验证：在请求到达控制器之前验证请求的数据格式和内容。

• 跨域资源共享（CORS）处理：处理跨域请求，允许或拒绝不同来源的请求。

• 国际化和本地化：根据用户的语言偏好和区域设置，提供对应语言和格式的响应内容。

• 限流：限制单个用户或IP的请求频率，防止滥用和DDoS攻击。

• 重定向和重写URL：根据请求的路径或参数进行URL重写或重定向。

• 身份验证：除了接口鉴权之外，还可以处理更复杂的身份验证流程，如OAuth、JWT等。

• 请求和响应转换：转换请求和响应的数据格式。

• 安全防护：检测和阻止恶意请求，如SQL注入、XSS攻击等

应用的生命周期事件

FastAPI 支持定义在应用启动前，或应用关闭后执行的事件处理器（函数）。可以很方便的在应用启动前初始化一些资源，关闭时释放一些资源。

之前装饰器写法

python

复制代码
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @Author: Hui
# @File: lifespan_demo.py
# @Desc: { fastapi应用生命周期事件Demo }
# @Date: 2024/06/25 17:43
import uvicorn
from fastapi import FastAPI

app = FastAPI()


async def init_setup():
    """初始化配置"""
    print("初始化日志、数据库...")


@app.on_event("startup")
async def startup():
    """应用启动前处理"""
    await init_setup()


@app.on_event("shutdown")
async def shutdown():
    """应用关闭时处理"""
    print("关闭数据库连接等...")
    print("Shutting down")


def main():
    uvicorn.run(app)


if __name__ == '__main__':
    main()

fastapi新推荐的上下文管理器写法

python

复制代码
import uvicorn
from fastapi import FastAPI
from contextlib import asynccontextmanager


async def init_setup():
    """初始化配置"""
    print("初始化日志、数据库...")


# @app.on_event("startup")
async def startup():
    """应用启动前处理"""
    await init_setup()


# @app.on_event("shutdown")
async def shutdown():
    """应用关闭时处理"""
    print("关闭数据库连接等...")
    print("Shutting down")
    

@asynccontextmanager
async def lifespan_demo(app: FastAPI):
    # 应用启动
    await startup()
    yield
    # 应用关闭
    await shutdown()


app = FastAPI(lifespan=lifespan_demo)


def main():
    uvicorn.run(app)


if __name__ == '__main__':
    main()

运行结果

源代码

Github：FastAPI 实战手册

从FastAPI的安装、请求处理、响应返回等基础知识入手，到中间件、数据库操作、身份认证等核心组件的应用，再到实战项目的开发，以及源码分析，循序渐进地介绍FastAPI的使用，旨在让读者全面掌握这个框架。

源文：FastAPI高级特性（二）：错误处理、中间件与生命周期

如有侵权请联系站点删除！