Denny的博客

Denny的博客

0%

FastAPI自学教程（31） - 子依赖项

发表于 更新于 分类于

1. 基础多级依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import Depends, FastAPI

app = FastAPI()

def query_extractor(q: str | None = None):
    return q

def query_or_body_extractor(
    q: str = Depends(query_extractor),
    body: dict | None = None
):
    return {"q": q, "body": body}

@app.post("/items/")
async def create_item(
    data: dict = Depends(query_or_body_extractor)
):
    return data
  • 依赖项可嵌套形成树状结构，query_or_body_extractor依赖query_extractor
  • 请求流程： 
      graph TD
  A[客户端请求] --> B[执行query_extractor]
  B --> C[执行query_or_body_extractor]
  C --> D[执行create_item]

2. 类型化依赖链

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pydantic import BaseModel

class UserAuth(BaseModel):
    username: str
    token: str

def get_token(authorization: str = Header(...)) -> str:
    return authorization.split()[-1]

def get_current_user(token: str = Depends(get_token)) -> UserAuth:
    return UserAuth(username=token[:5], token=token)

@app.get("/me")
async def user_profile(user: UserAuth = Depends(get_current_user)):
    return user
  • 类型系统自动执行三项核心功能：
    1. 请求解析：从Header提取Authorization字段
    2. 数据验证：验证JWT格式有效性
    3. 错误处理：自动生成401/403错误响应
  • 依赖链：get_token → get_current_user → user_profile

3. 缓存机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class Counter:
    def __init__(self):
        self.count = 0

def get_counter():
    return Counter()

def get_count(counter: Counter = Depends(get_counter)):
    counter.count += 1
    return counter.count

@app.get("/count")
async def show_count(
    a: int = Depends(get_count),
    b: int = Depends(get_count)
):
    return {"a": a, "b": b}
  • 同一请求中相同依赖项会被缓存复用（输出示例：{"a":1,"b":1}）
  • 使用use_cache=False禁用缓存： 
    1
    Depends(get_counter, use_cache=False)

4. 复杂依赖树

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def db_conn():
    return "DB_CONN"

def audit_log(conn: str = Depends(db_conn)):
    return f"Audit[{conn}]"

def security_check(
    conn: str = Depends(db_conn),
    log: str = Depends(audit_log)
):
    return f"Security[{conn}] with {log}"

@app.get("/admin")
async def admin_panel(
    check: str = Depends(security_check)
):
    return {"status": check}
  • 依赖树结构： 
      graph TD
  db_conn --> audit_log
  db_conn --> security_check
  audit_log --> security_check
  security_check --> admin_panel
  • 执行顺序：db_conn → audit_log → security_check

5. 生产实践建议

  1. 依赖隔离：业务逻辑依赖与基础设施依赖分层管理
  2. 性能监控：复杂依赖树需关注执行耗时
  3. 测试策略：使用FastAPI.TestClient模拟依赖项
  4. 错误追踪：为依赖项添加请求ID关联日志