Python 现代开发学习笔记

(十二)把数据库与身份接成依赖链

现在把示例中的普通资源换成 Session,并从请求头取得 Bearer 凭据。

# 项目文件:api_project/src/bookapi/dependencies.py
from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session

from .security import Identity, resolve_identity, unauthorized


def get_session(request: Request) -> Iterator[Session]:
    # 每个请求创建自己的 Session;with 离开时关闭并归还连接。
    with request.app.state.session_factory() as session:
        yield session


DbSession = Annotated[Session, Depends(get_session)]
bearer = HTTPBearer(auto_error=False)
BearerInput = Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]


def get_identity(session: DbSession, credentials: BearerInput) -> Identity:
    if credentials is None:
        raise unauthorized()
    return resolve_identity(session, credentials.credentials)


CurrentIdentity = Annotated[Identity, Depends(get_identity)]

Request 来自 FastAPI,它是 Starlette 请求对象的入口。request.app.state 保存应用启动时放入的工厂与配置。get_session 的 with 创建会话,yield 交给依赖者,收尾关闭会话;遇到异常也会离开 with。

HTTPBearer 来自 fastapi.security,负责解析请求头中的认证方案和凭据。auto_error=False 让缺少凭据时返回 None,由 get_identity 统一生成 401。它解析出文本,并不会自动确认令牌属于谁;查表与到期检查由 resolve_identity 完成。安全工具的参数可查 FastAPI HTTPBearer。

DbSession 和 CurrentIdentity 都是 Annotated 别名。路由写 session: DbSession、identity: CurrentIdentity,表达的是“请为本次请求提供这两项结果”。FastAPI 遇到 get_identity 又需要 get_session 时复用本次已有会话。会话只在这条请求内复用,其他请求取得自己的会话。

展开真实新增请求的依赖树,先取得 Session 和 Bearer 文本,再查令牌及用户,路由复用同一会话写入书籍,响应后关闭

笔记正文与全部题解 · Ctrl / ⌘ K