Python 完整项目阅读实战：从入口追到数据库

本篇目标

用一个 “生成笔记摘要” 的真实后端结构，把配置、FastAPI、依赖注入、Service、模型客户端、Repository、事务和测试串成完整调用链，并形成阅读陌生 Python 项目的固定方法。

读懂项目目录追踪请求调用链定位状态与副作用表达项目取舍

先记住一句话

阅读 Python 后端不要从每个文件逐行看，而要先找到启动入口和依赖组装，再选一条请求，从路由沿 Service、Repository 和外部客户端追到底，最后用测试确认真正契约。

项目场景与目录

这个项目提供 POST /notes/{id}/summary：验证当前用户后读取笔记，调用模型生成摘要，再把摘要写回数据库。

note-api/
├─ pyproject.toml                         # Python 版本、运行依赖和工具配置
├─ migrations/                            # Alembic 数据库迁移历史
├─ src/
│  └─ note_api/
│     ├─ main.py                          # 创建 FastAPI 应用并管理 lifespan
│     ├─ settings.py                      # 从环境读取配置
│     ├─ database.py                      # Engine、连接池和 Session 工厂
│     ├─ auth.py                          # 认证结果与请求身份
│     ├─ routes/
│     │  └─ summaries.py                  # HTTP 输入输出与错误映射
│     ├─ domain/
│     │  └─ summary.py                    # 用例、领域错误和所需接口
│     └─ infrastructure/
│        ├─ note_repository.py            # SQLAlchemy 持久化实现
│        └─ model_client.py                # 外部模型 HTTP 客户端
└─ tests/
   ├─ test_summary_service.py             # 使用 Fake 的业务单元测试
   └─ test_summaries_api.py               # HTTP 边界测试

目录不是重点，依赖方向才是重点：

main / routes / infrastructure
              ↓ 组装并实现
          domain service

domain 不导入 FastAPI、SQLAlchemy 或具体模型 SDK

这样业务用例可以在没有 HTTP、数据库和网络的情况下测试。项目规模很小时可以少建几层目录，但仍应保持 “业务规则不依赖具体基础设施” 的方向。

第一步：从启动命令找到入口

# 文件位置：pyproject.toml
[project]
name = "note-api"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
  "fastapi",               # HTTP 路由、依赖注入和 OpenAPI。
  "uvicorn[standard]",     # 运行 ASGI 应用的 Server。
  "httpx",                 # 异步调用外部模型服务。
  "sqlalchemy>=2,<3",      # ORM、Session 和事务。
  "asyncpg",               # PostgreSQL 的异步数据库驱动。
]

[project.scripts]
note-api = "note_api.cli:main" # 如果项目提供命令行入口，会从这里继续追踪。

常见部署命令是：

# note_api.main 表示模块，冒号后的 app 是模块中的 ASGI 应用对象。
uvicorn note_api.main:app --host 0.0.0.0 --port 8000

看到 uvicorn package.module:app，先打开对应模块，再找 FastAPI()、include_router() 和 lifespan。这比从名字猜哪个文件重要更可靠。

第二步：看入口怎样组装长期资源

# 文件位置：src/note_api/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI

from .database import engine
from .routes.summaries import router as summaries_router

# 装饰器把包含 yield 的异步生成器转换成 FastAPI 可用的生命周期管理器。
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # 每个应用进程创建一个 HTTP 连接池，供所有请求复用。
    app.state.model_http = httpx.AsyncClient(timeout=20.0)
    try:
        yield  # 资源准备完成后，应用才开始处理请求。
    finally:
        await app.state.model_http.aclose()
        await engine.dispose()  # 关闭阶段释放 HTTP 与数据库连接池。

def create_app() -> FastAPI:
    # 工厂函数集中组装应用；测试或不同环境可以分别创建实例。
    application = FastAPI(title="Note API", lifespan=lifespan)
    application.include_router(summaries_router)
    return application

app = create_app()  # Uvicorn 导入的就是这个对象。

这里要回答三个问题：资源何时创建、谁能访问、何时释放。这里的 Worker 不对应代码中的某个类或变量，而是 Uvicorn 启动的工作进程。例如增加 --workers 4 会启动 4 个 Worker；每个 Worker 都会独立导入这个模块、执行 app = create_app()，并各自运行一次 lifespan，因此它们拥有各自的应用对象、内存和连接池。

第三步：从路由只看输入、依赖和输出

# 文件位置：src/note_api/routes/summaries.py
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
from ..auth import Principal, require_principal
from ..database import session_factory
from ..domain.summary import NoteNotFound, SummaryService
from ..infrastructure.model_client import HTTPModelClient
from ..infrastructure.unit_of_work import SQLAlchemyUnitOfWork

# APIRouter 把同一前缀下的接口组织在一起，最后由入口统一挂载。
router = APIRouter(prefix="/notes", tags=["summaries"])

class SummaryResponse(BaseModel):
    # BaseModel 根据类型标注校验并序列化 HTTP 响应字段。
    note_id: str
    summary: str

def get_summary_service(
    request: Request,
) -> SummaryService:
    # request.app.state 读取 lifespan 启动时保存的共享 HTTP 客户端。
    model = HTTPModelClient(request.app.state.model_http)
    # 路由层只在组合位置选择具体实现，Service 只依赖协议。
    return SummaryService(
        # lambda 创建一个无参数工厂；每次调用都会得到新的工作单元。
        new_uow=lambda: SQLAlchemyUnitOfWork(session_factory),
        model=model,
    )

# 装饰器把函数注册为 POST 路由；{note_id} 会传给同名路径参数。
@router.post("/{note_id}/summary", response_model=SummaryResponse)
async def summarize_note(
    note_id: str,
    # Annotated 保留参数类型，并通过 Depends 告诉 FastAPI 怎样创建参数值。
    principal: Annotated[Principal, Depends(require_principal)],
    service: Annotated[SummaryService, Depends(get_summary_service)],
) -> SummaryResponse:
    try:
        summary = await service.summarize(note_id, principal.subject)
    except NoteNotFound as error:
        # 领域层不知道 HTTP；路由在边界把领域错误映射为状态码。
        raise HTTPException(status_code=404, detail="note not found") from error

    return SummaryResponse(note_id=note_id, summary=summary)

路由签名已经告诉你：路径参数是什么、身份从哪里来、Service 怎样组装、响应 Schema 是什么。此时不要立刻跳进框架源码，先进入 SummaryService.summarize() 看业务用例。

第四步：Service 展示真正业务顺序

# 文件位置：src/note_api/domain/summary.py
from typing import Protocol

# Protocol 只声明调用方需要的行为；具体实现不必显式继承这些协议。
class NoteNotFound(Exception):
    """笔记不存在，或当前用户没有访问权限。"""

class NoteRepository(Protocol):
    async def get_content(self, note_id: str, owner_id: str) -> str | None:
        """返回有权访问的笔记正文。"""
        ...  # Ellipsis 表示这里只声明契约，不提供方法实现。

    async def save_summary(self, note_id: str, owner_id: str, summary: str) -> None:
        """在当前事务中更新笔记摘要。"""
        ...  # 具体数据库写入由基础设施层实现。

class ModelClient(Protocol):
    async def summarize(self, content: str) -> str:
        """调用模型生成摘要。"""
        ...

class UnitOfWork(Protocol):
    notes: NoteRepository

    async def __aenter__(self) -> "UnitOfWork":
        # __aenter__ 和 __aexit__ 让对象可以用于 async with。
        """开启本次工作单元并提供 Repository。"""
        ...

    async def __aexit__(self, exc_type, exc, traceback) -> None:
        """根据是否出现异常提交或回滚，并释放 Session。"""
        ...

class UnitOfWorkFactory(Protocol):
    def __call__(self) -> UnitOfWork:
        # __call__ 让对象可以像函数一样调用，用于创建工作单元。
        """为一次短数据库交互创建新的工作单元。"""
        ...

class SummaryService:
    def __init__(
        self,
        new_uow: UnitOfWorkFactory,
        model: ModelClient,
    ) -> None:
        self._new_uow = new_uow
        self._model = model

    async def summarize(self, note_id: str, owner_id: str) -> str:
        # 读取使用独立的短工作单元，退出后立即归还数据库连接。
        async with self._new_uow() as read_uow:
            content = await read_uow.notes.get_content(note_id, owner_id)
        if content is None:
            raise NoteNotFound(note_id)

        # 模型调用发生在数据库事务之外，避免长时间占用连接和锁。
        summary = await self._model.summarize(content)

        # 写入使用新的短工作单元；成功退出时由 UoW 提交。
        async with self._new_uow() as write_uow:
            await write_uow.notes.save_summary(note_id, owner_id, summary)

        return summary

这段代码暴露了一个重要工程取舍：先读正文、释放查询使用的连接状态，再进行不确定耗时的模型调用，最后开启短事务写摘要。如果业务要求正文更新后旧模型结果不能覆盖新内容，还需要版本号或乐观锁，而不是简单扩大事务覆盖整个模型请求。

第五步：看基础设施如何实现协议

# 文件位置：src/note_api/infrastructure/note_repository.py
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import NoteModel

class SQLAlchemyNoteRepository:
    def __init__(self, session: AsyncSession) -> None:
        # Session 由工作单元创建并注入，Repository 不自行决定事务边界。
        self._session = session

    async def get_content(self, note_id: str, owner_id: str) -> str | None:
        # select() 和 where() 只构造 SQL 表达式，执行发生在 scalar()。
        statement = select(NoteModel.content).where(
            NoteModel.id == note_id,
            NoteModel.owner_id == owner_id,  # 资源权限进入查询条件。
        )
        return await self._session.scalar(statement)

    async def save_summary(self, note_id: str, owner_id: str, summary: str) -> None:
        # update().where().values() 构造带权限条件的更新语句。
        statement = (
            update(NoteModel)
            .where(NoteModel.id == note_id, NoteModel.owner_id == owner_id)
            .values(summary=summary)
        )
        result = await self._session.execute(statement)
        if result.rowcount != 1:
            # 更新目标消失或权限变化时，不允许静默成功。
            raise RuntimeError("note changed before summary was saved")

# 文件位置：src/note_api/infrastructure/unit_of_work.py
from types import TracebackType

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from .note_repository import SQLAlchemyNoteRepository

class SQLAlchemyUnitOfWork:
    def __init__(self, factory: async_sessionmaker[AsyncSession]) -> None:
        # factory 是 Session 工厂；真正的 Session 到进入工作单元时才创建。
        self._factory = factory
        self._session: AsyncSession | None = None
        self.notes: SQLAlchemyNoteRepository

    async def __aenter__(self) -> "SQLAlchemyUnitOfWork":
        # async with 进入时自动调用 __aenter__，返回值绑定给 as 后的变量。
        self._session = self._factory()
        await self._session.begin()  # 当前工作单元使用一个明确事务。
        self.notes = SQLAlchemyNoteRepository(self._session)
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        # assert 表达内部不变量，并帮助类型检查器确认后续不再是 None。
        assert self._session is not None
        try:
            if exc is None:
                await self._session.commit()  # 整个工作单元成功才提交。
            else:
                await self._session.rollback()  # 任意异常都会回滚。
        finally:
            await self._session.close()  # 无论提交或回滚是否成功都释放资源。

# 文件位置：src/note_api/infrastructure/model_client.py
import httpx

class HTTPModelClient:
    def __init__(self, client: httpx.AsyncClient) -> None:
        # 客户端由应用生命周期创建并注入，以复用连接池。
        self._client = client

    async def summarize(self, content: str) -> str:
        # await 等待异步 HTTP 请求；json 参数会被编码为 JSON 请求体。
        response = await self._client.post(
            "http://model-service.internal/summarize",
            json={"content": content},
        )
        response.raise_for_status()  # 非 2xx 响应转成明确异常。
        # 外部响应仍是不可信数据，解析 JSON 后继续检查字段类型和内容。
        payload = response.json()
        summary = payload.get("summary")
        if not isinstance(summary, str) or not summary.strip():
            raise ValueError("model response has no valid summary")
        return summary.strip()

模型 URL 在完整项目中应来自 Settings，而不是硬编码；这里保留固定内部地址只是为了让代码片段聚焦调用边界。对外部响应做结构校验很重要，因为 HTTP 200 不等于业务字段一定正确。

第六步：用测试确认契约，而不是猜实现

# 文件位置：tests/test_summary_service.py
from dataclasses import dataclass

import pytest

from note_api.domain.summary import NoteNotFound, SummaryService

class FakeNotes:
    # Fake 是测试替身，只在内存中记录状态，不连接真实数据库。
    def __init__(self, content: str | None) -> None:
        self.content = content
        self.saved_summary: str | None = None

    async def get_content(self, note_id: str, owner_id: str) -> str | None:
        return self.content

    async def save_summary(self, note_id: str, owner_id: str, summary: str) -> None:
        self.saved_summary = summary

# @dataclass 为测试数据对象自动生成初始化方法，减少无关样板代码。
@dataclass
class FakeUnitOfWork:
    notes: FakeNotes
    committed: bool = False

    async def __aenter__(self) -> "FakeUnitOfWork":
        return self

    async def __aexit__(self, exc_type, exc, traceback) -> None:
        self.committed = exc is None  # 无异常退出代表工作单元成功。

class FakeModel:
    def __init__(self) -> None:
        self.calls = 0

    async def summarize(self, content: str) -> str:
        self.calls += 1  # 记录调用次数，用来验证失败路径没有产生模型费用。
        return f"摘要：{content}"

# pytest.mark.asyncio 让 pytest 在事件循环中执行异步测试函数。
@pytest.mark.asyncio
async def test_summary_is_persisted() -> None:
    notes = FakeNotes(content="Python 运行时")
    work_units: list[FakeUnitOfWork] = []

    def new_uow() -> FakeUnitOfWork:
        # 嵌套工厂会记住外层的 notes 和 work_units，这里利用了闭包。
        unit = FakeUnitOfWork(notes=notes)
        work_units.append(unit)
        return unit

    service = SummaryService(new_uow=new_uow, model=FakeModel())

    result = await service.summarize("note-1", "user-1")

    assert result == "摘要：Python 运行时"
    assert notes.saved_summary == result
    assert len(work_units) == 2  # 读取和写入没有跨模型调用占用同一事务。
    assert all(unit.committed for unit in work_units)

@pytest.mark.asyncio
async def test_missing_note_does_not_call_model() -> None:
    notes = FakeNotes(content=None)
    model = FakeModel()
    service = SummaryService(
        new_uow=lambda: FakeUnitOfWork(notes=notes),
        model=model,
    )

    # pytest.raises 断言代码块必须抛出指定异常，否则测试失败。
    with pytest.raises(NoteNotFound):
        await service.summarize("missing", "user-1")

    assert model.calls == 0  # 笔记不存在时必须在调用付费模型前停止。

这里的 Service 只依赖 Unit of Work 协议，测试不需要模拟 SQLAlchemy。读取和写入各自拥有短事务，模型等待期间不会占用数据库连接；如果一个业务步骤要同时修改多张表，则在同一个写工作单元中调用多个 Repository。

阅读陌生项目的九步法

1. 看 README、pyproject.toml、容器文件和启动命令；
2. 找 ASGI / CLI / Worker 入口，不先逐个目录浏览；
3. 看配置从环境、文件还是远端配置中心进入；
4. 看入口怎样创建连接池、客户端和后台任务；
5. 选一条关键路由或消息消费入口；
6. 沿 Service 读业务顺序、事务边界和领域错误；
7. 沿 Repository、模型客户端或消息客户端读副作用；
8. 查测试确认稳定契约和重要边界；
9. 最后再看框架封装、公共工具和不在主链路上的文件。

遇到装饰器或依赖注入时，先写出 “谁调用谁” 的展开版：

FastAPI 收到请求
├─ 调 require_principal()
├─ 调 get_session() 并获得 AsyncSession
├─ 调 get_summary_service() 组装具体实现
└─ 调 summarize_note()
   └─ 调 SummaryService.summarize()
      ├─ Repository.get_content()
      ├─ ModelClient.summarize()
      └─ Repository.save_summary()

项目面试怎样表达

不要只说 “我用了 FastAPI、SQLAlchemy 和 asyncio” ，而要说清问题、边界和验证：

我把摘要接口拆成 HTTP、业务用例和基础设施三层。FastAPI 负责校验和依赖组装，Service 只依赖 Repository 与模型客户端协议。模型调用不放在数据库事务中，避免长事务占用连接；写回阶段使用短事务，并通过用户条件限制资源访问。测试方面，单元测试注入 Fake 验证业务顺序，数据库和路由另做集成测试，模型质量由独立评测集覆盖。

面试官继续追问时，应能回答：

为什么 AsyncSession 不能跨并发 Task 共享？参考答案

AsyncSession 是可变且有状态的工作单元，会跟踪 ORM 对象、待写入变化、事务和连接。多个并发 Task 共用时，操作顺序和事务状态可能互相干扰；应该让每个 Task 使用独立 Session，只共享线程安全的 Engine 与连接池。

模型超时或用户断开后，怎样取消并防止写回？参考答案

模型客户端要设置超时，并让取消信号继续传递到下游调用。检测到超时或断开后，应取消并等待任务完成清理；只有模型成功返回且任务仍有效时才写数据库，再结合版本条件或幂等键阻止迟到结果覆盖新数据。

怎样用版本字段避免旧摘要覆盖新正文？参考答案

读取正文时同时记录当前版本号，模型完成后使用带版本条件的更新，例如只在 version = original_version 时写入摘要。如果受影响行数为 0，说明正文已经变化，应丢弃旧结果或基于新正文重新生成。

为什么认证成功后仍要做资源级授权？参考答案

认证只能证明用户是谁，不能证明他可以访问某条笔记。查询和更新时仍要把 owner_id、组织或角色等权限条件放进数据访问边界，避免用户仅修改资源 ID 就读取或覆盖他人的数据。

怎样观察模型延迟、数据库连接等待和错误率？参考答案

分别记录模型请求耗时、超时与状态，连接池等待时间、活跃连接与事务时长，以及 HTTP 状态码和错误率。再用同一个请求 ID 或 Trace 串联一次调用，设置延迟和错误率告警，才能判断瓶颈在模型、数据库还是应用代码。

什么时候应该把同步摘要改成任务队列？参考答案

当模型调用经常超过 HTTP 等待时间，或者任务需要重试、限流、批处理并在用户断开后继续执行时，应改用任务队列。接口只创建任务并返回任务 ID，队列消费者异步处理，客户端再查询状态或接收完成通知；写回操作还要保证幂等。

高频面试题与回答

1. 你接手一个陌生 FastAPI 项目会从哪里开始看？参考答案

先从启动命令找到应用入口，看 lifespan 和路由注册，确认全局资源怎样组装。然后选择一条核心请求，沿路由、依赖、Service、Repository 和外部客户端追踪，再通过测试、Schema 和迁移确认契约，不会从所有工具文件逐行开始。

2. 为什么模型调用不放进数据库事务？参考答案

模型调用延迟高且可能重试或超时，如果事务一直打开，会长期占用连接和锁，放大数据库压力。通常先完成外部计算，再用短事务写结果；若需要防止数据变化，可以使用版本号、幂等键或 Outbox 等明确机制。

3. 分层是否意味着每个项目都要建很多目录？参考答案

不是。分层的核心是职责与依赖方向，而不是目录数量。小项目可以把文件放得更近，但业务规则不应直接依赖 HTTP 或具体数据库，副作用边界要能替换和测试；只有复杂度真实增长时再拆目录。

接下来学什么

完成这条调用链后，可以选择一个你实际参与的 TypeScript 后端，用同样九步法分别画出 Python 与原项目的入口、依赖和事务边界。下一模块进入 Go 开发总览，比较编译型服务的项目阅读方式。

参考资料

• Python Packaging：pyproject.toml (opens new window)
• FastAPI 大型应用结构 (opens new window)
• SQLAlchemy asyncio (opens new window)
• HTTPX 异步客户端 (opens new window)