（五）FastAPI的依赖注入系统

FastAPI 的依赖注入系统是其最强大和最灵活的特性之一，使得代码更模块化、更易维护和更易测试。本文将详细介绍 FastAPI 的依赖注入系统，包括基本用法、高级用法和实际应用场景。

1. 什么是依赖注入

依赖注入（Dependency Injection, DI）是一种设计模式，它通过外部注入而不是内部创建的方式，将对象的依赖关系传递给对象。这样可以提高代码的可测试性和可维护性。FastAPI 使用Depends类来实现依赖注入。

2. 基本用法

在 FastAPI 中，可以使用 Depends 声明路径操作函数的依赖项。依赖项可以是函数、类或者其他对象，FastAPI 会自动解析这些依赖项，并将它们的返回值作为参数传递给路径操作函数。

from fastapi import Depends, FastAPI

app = FastAPI()

def common_parameters(q: str = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

在这个示例中，common_parameters 是一个依赖项，定义了一组常用的查询参数。路径操作函数 read_items 使用 Depends 引入这些参数。

3. 高级用法

3.1 类作为依赖项

依赖项不仅可以是函数，还可以是类实例。这在需要维护状态或者初始化复杂逻辑时非常有用。

from fastapi import Depends, FastAPI

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 10):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

3.2 子依赖项

依赖项本身可以有依赖项，形成依赖链。FastAPI 会自动解析整个依赖链，并注入所需的参数。

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    return x_token

def verify_key(x_key: str = Header(...), token: str = Depends(verify_token)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="Invalid X-Key header")
    return x_key

@app.get("/items/")
async def read_items(key: str = Depends(verify_key)):
    return {"key": key}

在这个示例中，verify_key 依赖 verify_token，形成一个依赖链。FastAPI 会首先解析 verify_token，然后解析 verify_key，最后将 key 注入到路径操作函数 read_items 中。

4. 依赖注入的实际应用场景

4.1 认证和授权

依赖注入在实现认证和授权时非常有用。你可以定义一个认证依赖项，在所有需要认证的路径操作函数中复用它。

from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "fake-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "fake-user"}

@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

4.2 数据库连接

在 Web 应用中，处理数据库连接是一个常见的需求。你可以使用依赖注入来管理数据库连接的生命周期。

from fastapi import Depends, FastAPI
import databases

app = FastAPI()

DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)


async def get_db():
    await database.connect()
    print('database connect.')
    try:
        yield database
    finally:
        await database.disconnect()
        print('database disconnect.')


@app.get("/items/")
async def read_items(db: databases.Database = Depends(get_db)):
    query = "SELECT * FROM items"
    response = await db.fetch_all(query)
    print('database execute query.')
    return response

4.3 日志记录

你可以使用依赖注入在每个请求中添加日志记录功能。

from fastapi import Depends, FastAPI, Request
import logging

app = FastAPI()

logging.basicConfig(level=logging.INFO)

async def log_request(request: Request):
    logging.info(f"Request: {request.method} {request.url}")
    return request

@app.get("/items/")
async def read_items(request: Request = Depends(log_request)):
    return {"message": "Check the logs for the request details"}

5. 总结

FastAPI 的依赖注入系统提供了一种强大且灵活的方式来管理路径操作函数的依赖项。通过 Depends，你可以将常用的逻辑抽象为依赖项，并在多个路径操作函数中复用。无论是处理认证和授权、管理数据库连接，还是添加日志记录，依赖注入都能帮助你编写更加模块化、可维护和可测试的代码。

希望这篇文章能帮助你深入理解 FastAPI 的依赖注入系统。如果你有任何问题或需要进一步的帮助，请随时联系我。