FastAPI基础：6.中间件&依赖注入

中间件

ps：访问流程大致如下：进入中间件 - 视图 - 出中间件 - 客户端

什么是跨域？如何实现（详解） - 知乎 (zhihu.com)

from typing import Union

from fastapi import FastAPI, HTTPException
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from uvicorn import run
from fastapi_04 import R

"""
中间件 & 依赖
"""
from fastapi.middleware.cors import CORSMiddleware

# 自定义中间件
# 2023-07-01 建议使用route_class实现请求响应日志记录：
# https://github.com/zy7y/mini-rbac/commit/0ef41c6cdecf18c444010dd6fdc14e5c3a015728
class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print(" CustomHeaderMiddleware 收到了请求-到达中间件，处理请求逻辑开始")
        response = await call_next(request)
        print("CustomHeaderMiddleware 执行了视图函数，拿到响应数据，处理响应数据")
        response.headers['Custom'] = 'Example'
        print("CustomHeaderMiddleware 处理完了 返回")
        return response

middleware = [
    # 跨域中间件
    Middleware(CORSMiddleware, allow_origins=['*']),
    Middleware(CustomHeaderMiddleware) # 最先添加的中间件
]
# 组册中间件的方式1
app = FastAPI(middleware=middleware)

# 一个允许跨域请求的源列表 ['*'] 代表所有
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],  # 一个允许跨域请求的 HTTP 方法列表
    allow_headers=["*"],  # 一个允许跨域请求的 HTTP 请求头列表
)

# 注册中间件方式2
@app.middleware("http")  # 最后添加的中间件
async def globalExceptionHandler(request: Request, call_next):
    try:
        print(" globalExceptionHandler 检查是否存在异常")
        response = await call_next(request)
    except HTTPException as e:
        print(" globalExceptionHandler 返会异常对象")
        return R.fail(str(e))
    print("globalExceptionHandler 无异常返回")
    return response

@app.get("/")
async def main():
    return {"message": "Hello World"}

if __name__ == "__main__":
    # 启用web服务， reload -> 当改变代码时自动重启 # fastapi_02:app
    run("__main__:app", reload=True)

ps：访问main打印信息

globalExceptionHandler 检查是否存在异常。# 先执行最后添加中间见的call-next前面部分
CustomHeaderMiddleware 收到了请求-到达中间件，处理请求逻辑开始# 执行最先添加的 call-next前面部分
CustomHeaderMiddleware 执行了视图函数，拿到响应数据，处理响应数据 # 执行完后执行最先添加的 call-next后面部分
CustomHeaderMiddleware 处理完了 返回
globalExceptionHandler 无异常返回 # 最后执行最后添加的中间件call-next后面部分

依赖注入

ps： 多种形式, 个人感觉和中间件有点类似

# 依赖注入
def depends_hello():
    print("depends_hello , run")
    yield 1
    print("depends_hello, over")

# 类依赖
class CommonQueryParams:
    def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

from fastapi import APIRouter, Depends

# 该router下所有路由都会执行这个 依赖； 全局依赖注入、
router = APIRouter(dependencies=[Depends(CommonQueryParams)])

@router.get("/123")
async def hello123():
    return 123

# 对这个视图注入，可以拿到参数
@router.get("/456")
async def hello456(num: int = Depends(depends_hello)):
    print("进了请求")
    return num

app.include_router(router, dependencies=[])

@app.get("/321")
async def hello321(num = Depends(CommonQueryParams)):
    return num

完整代码

from typing import Union

from fastapi import FastAPI, HTTPException
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from uvicorn import run
from fastapi_04 import R

"""
中间件 & 依赖
"""
from fastapi.middleware.cors import CORSMiddleware

# 自定义中间件
class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print(" CustomHeaderMiddleware 收到了请求-到达中间件，处理请求逻辑开始")
        response = await call_next(request)
        print("CustomHeaderMiddleware 执行了视图函数，拿到响应数据，处理响应数据")
        response.headers['Custom'] = 'Example'
        print("CustomHeaderMiddleware 处理完了 返回")
        return response

middleware = [
    # 跨域中间件
    Middleware(CORSMiddleware, allow_origins=['*']),
    Middleware(CustomHeaderMiddleware)
]

app = FastAPI(middleware=middleware)

# 一个允许跨域请求的源列表 ['*'] 代表所有
origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],  # 一个允许跨域请求的 HTTP 方法列表
    allow_headers=["*"],  # 一个允许跨域请求的 HTTP 请求头列表
)

@app.middleware("http")
async def globalExceptionHandler(request: Request, call_next):
    try:
        print(" globalExceptionHandler 检查是否存在异常")
        response = await call_next(request)
    except HTTPException as e:
        print(" globalExceptionHandler 返会异常对象")
        return R.fail(str(e))
    print("globalExceptionHandler 无异常返回")
    return response

@app.get("/")
async def main():
    return {"message": "Hello World"}

@app.get("/ok")
async def ok():
    raise HTTPException(status_code=401)

    # 依赖注入
def depends_hello():
    print("depends_hello , run")
    yield 1
    print("depends_hello, over")

    # 类依赖
class CommonQueryParams:
    def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

from fastapi import APIRouter, Depends

# 该router下所有路由都会执行这个 依赖； 全局依赖注入、
router = APIRouter(dependencies=[Depends(CommonQueryParams)])

@router.get("/123")
async def hello123():
    return 123

 # 拿到参数， 对这个视图注入，可以拿到参数，num 是注入依赖的调用结果，不会被作为查询参数
@router.get("/456")
async def hello456(num: int = Depends(depends_hello)):
    print("进了请求")
    return num

app.include_router(router, dependencies=[])

@app.get("/321")
async def hello321(num = Depends(CommonQueryParams)):
    return num

if __name__ == "__main__":
    # 启用web服务， reload -> 当改变代码时自动重启 # fastapi_02:app
    run("__main__:app", reload=True)

ps：访问 /hello456 打印

globalExceptionHandler 检查是否存在异常
CustomHeaderMiddleware 收到了请求-到达中间件，处理请求逻辑开始
depends_hello , run
进了请求
CustomHeaderMiddleware 执行了视图函数，拿到响应数据，处理响应数据
CustomHeaderMiddleware 处理完了 返回
globalExceptionHandler 无异常返回
INFO:     127.0.0.1:50879 - "GET /456?q=123&skip=0&limit=100 HTTP/1.1" 200 OK
depends_hello, over