ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

FastAPI项目结构演进:从单文件到工程化拆分最佳实践

FastAPI项目结构演进:从单文件到工程化拆分最佳实践 我刚接触 FastAPI 的时候所有代码都在 main.py 里路由、数据库连接、ORM 模型、Pydantic schema、业务逻辑全堆在一起。小项目跑得确实快等路由超过二十个、测试想连临时数据库都要改全局变量、同事之间改代码总是冲突的时候我才意识到真正值钱的不是什么优雅框架而是 FastAPI 项目结构本身——结构承担的是沟通成本、测试成本和变更成本。这篇不是给你一份放之四海皆准的模板而是把我自己从单文件到工程化结构的演进过程完整写出来适合刚写完第一个接口的人也适合正在为目录越来越乱而头疼的团队。1. 先接受现实单文件能跑但跑不远FastAPI 的第一个特点就是“快”这个快不只是性能还有上手速度。你只需要十行代码就能起一个带 Swagger 文档的服务于是绝大多数项目都会长成这样project/ ├── main.py # FastAPI 实例和全部路由 ├── database.py # engine、SessionLocal ├── models.py # 所有 ORM 模型 └── schemas.py # 所有 Pydantic 模型这种结构在接口数量小于十个、业务逻辑只有增删改查的时候没有任何问题。但我见过不少项目在 main.py 写到两千行之后仍然没有拆分所有人都知道该拆但没人敢动。为什么因为耦合已经变成了一张网改一个路由函数要担心另一个 import 它改一个 model 字段要担心所有接口的返回结构。1.1 两千行 main.py 是怎么把我逼到拆分的真正让我下决心拆分的不是代码行数而是三个具体症状。第一个症状是“加一个接口需要看完整份文件”。新需求通常很简单比如给用户列表加一个分页参数但滚到对应函数之前你已经被前面的各种依赖、中间件、异常处理分散了注意力。改动本身不需要一分钟找到正确的位置却要十分钟。第二个症状是测试无法独立执行。写单元测试时想用内存数据库替换生产数据库结果database.py里 engine 在 import 阶段就初始化了测试进程一开始就连接了真实数据库。想 mock 某个 service又发现业务逻辑没有通过依赖注入而是直接 import 模块内的全局函数魔改的成本非常高。第三个症状更隐蔽代码 review 越来越难。当路由、服务、模型混在一个文件里时reviewer 无法一目了然地判断某次改动到底是“新增接口”还是“调整数据层”。review 的关注点一旦分散质量问题就会被漏掉。拆分的本质不是“好看”而是让每次改动的定位范围尽量小让每个模块可以被单独替换、单独测试。1.2 第一次调整先按技术层拆还是先按业务拆拆目录时最常见的纠结是横向按技术层拆还是纵向按业务域拆。这两种思路我在不同项目里都试过。拆分方式适合阶段优势风险横向分层api / models / services接口少、业务单一概念直观新手容易理解业务增加后同一业务的代码散到多个目录纵向业务域user / order / payment业务变多、多人协作改一个业务时集中在一个目录初期容易过度设计公共代码需要整理我的建议是按顺序来项目刚开始可以先横向分但一旦出现第二个真正的业务域立刻切换成纵向结构。判断“真正的业务域”有一个简单标准这个模块有没有自己独立的表、独立的接口路径、独立的权限规则。如果都有它就配拥有一个自己的目录。2. 中型项目推荐结构业务域模块 依赖注入当业务开始变多我推荐用下面这套目录。它不是src / tests那种教科书结构而是我在真实项目里跑过一段时间的结构兼顾了 FastAPI 的依赖注入特性和团队协作的清晰度。project/ ├── app/ │ ├── main.py │ ├── core/ │ │ ├── config.py # Pydantic Settings │ │ ├── security.py # 密码哈希、JWT │ │ └── deps.py # 公共依赖get_current_user 等 │ ├── db/ │ │ ├── base.py # 模型基类、公共字段 │ │ └── session.py # engine、SessionLocal、get_db │ ├── modules/ │ │ ├── user/ │ │ │ ├── router.py # HTTP 路由 │ │ │ ├── schemas.py # Pydantic 输入输出模型 │ │ │ ├── models.py # SQLAlchemy ORM 模型 │ │ │ ├── service.py # 业务规则 │ │ │ └── repository.py # 数据访问层 │ │ └── order/ │ │ ├── router.py │ │ ├── schemas.py │ │ ├── models.py │ │ ├── service.py │ │ └── repository.py │ ├── common/ │ │ ├── exceptions.py # 自定义异常 │ │ ├── response.py # 统一响应结构 │ │ └── pagination.py # 分页参数依赖 │ └── tests/ ├── alembic/ ├── .env.example ├── pyproject.toml └── Dockerfilecore放横切关注点modules放业务域common放被多个模块共享的工具。核心原则是一个业务域的代码路径尽量短从一个入口进去就能看全这个业务的输入输出、业务规则和数据访问。2.1 业务域模块的职责边界很多人第一次看到user目录里又有 service 又有 repository 时会觉得过度设计。这里的关键是看你的业务到底复不复杂。我的判断标准很朴素如果业务规则只是“查出来返回”那 service 和 repository 完全可以合并成一个 service.py如果业务规则已经涉及到状态流转、金额计算、权限判断那 service 单独存在就是值得的。repository 的引入时机更晚只有当同一份查询逻辑需要在多个 service 里复用或者你需要在 service 测试中替换数据源的时候才给它位置。在目录设计层面router.py应该是最薄的一层。它只负责 HTTP 语义接收参数、调用 service、返回 response_model 指定的结构。不要在 router 里写业务规则否则你迟早会遇到“两个接口复制粘贴了同一段逻辑”的尴尬。2.2 看一个用户的完整请求链路下面我用用户模块举个例子。路由层长这样# app/modules/user/router.py from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.db.session import get_db from app.modules.user import schemas, service router APIRouter(prefix/users, tags[users]) router.get(/{user_id}, response_modelschemas.UserOut) def read_user(user_id: int, db: Session Depends(get_db)): user service.get_user(db, user_id) if not user: raise HTTPException(status_code404, detailuser not found) return userPydantic schema 单独放# app/modules/user/schemas.py from pydantic import BaseModel, ConfigDict class UserOut(BaseModel): id: int name: str email: str model_config ConfigDict(from_attributesTrue)service 层管理业务规则和数据访问。如果这个查询不复杂可以合并 repository 的职责# app/modules/user/service.py from sqlalchemy.orm import Session from app.modules.user.models import User def get_user(db: Session, user_id: int) - User | None: return db.get(User, user_id)三个文件都有自己的职责router 决定输入输出和状态码schemas 约束外部契约service 组织业务逻辑和数据访问。模块之间不直接相互依赖具体实现而是通过依赖注入把 db 传递进来。2.3 依赖注入是模块之间唯一的“接头暗号”FastAPI 的Depends是我认为整个框架里最容易被低估的机制。很多人只是把它当作获取请求参数的方式实际上它是模块之间的接线协议。看这段典型代码# app/core/deps.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from app.db.session import get_db from app.modules.user.models import User oauth2_scheme OAuth2PasswordBearer(tokenUrl/api/v1/auth/login) def get_current_user( token: str Depends(oauth2_scheme), dbDepends(get_db), ) - User: ...任何路由只要写current_user: User Depends(get_current_user)就能拿到当前登录用户而路由自己完全不知道 token 是从哪来、怎么解析的。这种解耦让测试变得非常舒服FastAPI 提供app.dependency_overrides测试环境直接替换掉get_db或get_current_user业务模块不需要任何改动。3. 横切关注点要单独安置配置、数据库会话、异常处理业务模块再清楚如果配置、数据库会话、异常处理这些横切关注点散落在各处结构还是会崩。它们是所有模块都要依赖的地基必须在早期就固定住位置。3.1 Settings 对象把 .env 变成类型安全的配置入口我踩过的坑是项目里到处读os.environ.get(DATABASE_URL)一旦有人把变量名写错页面接口就静默失败。后来统一改成 Pydantic Settings# app/core/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): app_name: str demo environment: str development database_url: str sqlite:///./dev.db secret_key: str change-me model_config SettingsConfigDict(env_file.env, extraignore) settings Settings()这样做的收益非常明显配置字段有类型校验IDE 能自动补全测试时想覆盖配置只需要在创建测试进程前设置环境变量.env.example可以提供一份完整的配置样例新人照着复制就能跑。如果你在上线后看到“默认值生效了但线上配置没生效”之类的诡异问题多半是环境变量名和 Settings 字段名对不上。3.2 数据库 Session 的 yield 依赖每个请求各用各的数据库会话是新手最容易写错的部分。最典型的错误是把SessionLocal()放在模块顶部做成一个全局 session然后所有路由共享它。这在并发场景下很快就会出问题一个请求修改了 session 状态另一个请求复用同一个 session数据读到一半被其他请求提交破坏。正确的做法是用 yield 依赖让 FastAPI 在每个请求的生命周期里创建一个新 session# app/db/session.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.core.config import settings engine create_engine(settings.database_url, pool_pre_pingTrue) SessionLocal sessionmaker(bindengine, autocommitFalse, autoflushFalse) def get_db(): db SessionLocal() try: yield db finally: db.close()注意这里用的是yield而不是return因为finally里的db.close()必须在每个请求结束后执行。FastAPI 对这种 yield 依赖有完整的清理机制这也是为什么不要在 service 里手动创建 session 的原因一旦手动创建你就必须手动管理关闭时机。如果你用的是async版 SQLAlchemy思路完全一样只是把create_engine换成create_async_engine把sessionmaker换成async_sessionmaker然后把依赖定义成 async generator。3.3 中间件和异常处理器该放在哪一层中间件和异常处理器和具体业务没有关系但它们影响着所有接口。我的做法是放在 main.py 的注册阶段而不是塞在某个 module 里。# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.core.config import settings from app.common.exceptions import register_exception_handlers from app.modules.user.router import router as user_router def create_app() - FastAPI: app FastAPI(titlesettings.app_name) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) register_exception_handlers(app) app.include_router(user_router, prefix/api/v1) return app app create_app()一个容易被忽略的细节是异常处理器的注册顺序。FastAPI 在路由抛出异常时会按注册顺序寻找匹配的 handler如果你在中间件里做全局 try except再注册 handler错误信息可能会被中间件吞掉。我建议把异常处理集中放在common/exceptions.py里main.py 只调用注册函数避免 main.py 变成一颗“什么都会”的上帝文件。4. 复杂项目的升级路径应用工厂、仓储层和边界规则当团队、测试、部署环境都开始变复杂时光有 business-domain 目录还不够。你需要回头审视 FastAPI 实例本身是怎么创建的。4.1 应用工厂为测试和部署提供干净的 FastAPI 实例最开始的 main.py 里app FastAPI()是全局的。这对只跑一个环境来说没问题但你迟早会碰到三种场景测试需要用一个不同的 application 实例本地、预发布、生产需要不同的配置你要写多个启动入口比如 API 服务和定时任务服务。这时候应该引入应用工厂模式# app/main.py from fastapi import FastAPI from app.core.config import settings def create_app() - FastAPI: app FastAPI(titlesettings.app_name) register_database(app) register_routers(app) register_exception_handlers(app) return app app create_app()测试时的用法变成了app create_app() app.dependency_overrides[get_db] override_get_db client TestClient(app)工厂模式的本质是把“创建应用”变成一个有输入输出的函数而不是一个脚本。部署时 uvicorn 依然认app.main:app这个变量就是工厂创建出的实例测试时你则可以随时创建任意配置的实例互不污染。4.2 Repository/Service 模式别为了设计感而用到了更大规模的项目很多人会引入 Repository 层。它的作用是隔离 SQLAlchemy 的查询细节# app/modules/user/repository.py from sqlalchemy.orm import Session from app.modules.user.models import User class UserRepository: def __init__(self, db: Session): self.db db def get(self, user_id: int) - User | None: return self.db.get(User, user_id) def find_by_email(self, email: str) - User | None: return self.db.query(User).filter(User.email email).first()service 只依赖 repository 接口不依赖数据库知识# app/modules/user/service.py from app.modules.user.repository import UserRepository class UserService: def __init__(self, repo: UserRepository): self.repo repo def get_user(self, user_id: int): return self.repo.get(user_id)但在决定用它的第一天你必须回答一个问题这个抽象为我缓存了什么如果只是把db.query(User).filter(User.id user_id).first()换成repo.get(user_id)那这层抽象是负债不是资产。只有当你真正需要在 service 测试中用内存 mock 替换数据库或者同一套查询逻辑要被多个 service 复用、未来还可能接 Redis 缓存时Repository 模式才有价值。我见过太多项目为了“整洁架构”而建了五层目录结果每一层都只是透传最后改一个查询条件要摸五个文件。4.3 API 版本化、后台任务和大文件处理在结构上的落位API 版本化最直接的做法是给include_router加前缀。现阶段app.include_router(user_router, prefix/api/v1)将来上线 v2 时不需要把旧路由全部搬走而是在app/api/v2里建新路由再 include 一次。版本化是外部契约模块内部业务不用为此重写。后台任务我建议分成两个量级。轻量、只在本进程执行的直接使用 FastAPI 自带的BackgroundTasks涉及到定时、跨进程、需要重试的任务不要塞进 FastAPI app 里单独建app/modules/xxx/tasks.py交给 Celery 或 ARQ 这类任务队列。目录上保持“模块内 tasks.py 模块外 task worker”的结构比把所有任务都堆到app/tasks.py要清晰得多。大文件上传也是一样的思路不要把文件流处理的代码写在路由函数里。路由只接收UploadFile然后把文件交给 service 或独立的上传模块。目录结构只是载体真正重要的是每个非路由进程能独立找到自己的代码。4.4 拆成微服务的信号藏在哪些业务矛盾里很多团队在项目还不到一万行时就开始计划拆分微服务我觉得这是本末倒置。一个 FastAPI 单体服务撑到相当大的用户量是完全可行的真正的拆服信号往往来自组织和流程某个子域需要独立部署否则上线要跟着其他模块一起等某个子域需要不同的扩缩容策略两个团队同时改同一个仓库导致合并成本暴涨。但在拆之前你必须先把单体内的业务边界画清楚。如果在一个 monorepo 里你都无法让 user 模块不依赖 order 模块的内部实现拆成微服务只会把代码内耦合变成网络耦合。架构升级是结构问题的解决手段不是逃避当前混乱的捷径。5. 测试、打包与部署让结构在真实环境里成立再漂亮的目录结构如果测试没法跑、部署起来日志混乱那也只是空架子。这一节聊聊真正用过之后才会注意到的细节。5.1 测试目录怎么布局依赖覆盖怎么写测试目录最好镜像业务模块tests/ ├── conftest.py ├── api/ │ └── test_users.py ├── services/ │ └── test_user_service.py └── factories.pyconftest 里最关键的是覆盖 get_db 依赖。用内存 SQLite 或者独立测试库替换真实数据库# tests/conftest.py import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.db.session import get_db from app.main import create_app engine create_engine( sqlite:///./test.db, connect_args{check_same_thread: False}, ) TestingSession sessionmaker(bindengine, autocommitFalse, autoflushFalse) pytest.fixture() def client(): app create_app() def override_get_db(): db TestingSession() try: yield db finally: db.close() app.dependency_overrides[get_db] override_get_db return TestClient(app)这里的关键是用create_app()而不是直接 import 全局app这样每次测试都能拿到一个干净的实例。覆盖get_db后所有路由里的Depends(get_db)自动走测试数据库业务模块一个字节都不用改。5.2 从开发机到 Docker启动方式、worker 数量与健康检查部署层面的结构最容易被忽略。我的建议是开发环境用--reload生产环境永远不要开 reload# 开发 uvicorn app.main:app --reload # 生产 uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 2--reload会监听文件变化并自动重启进程生产环境一旦有人误触发了文件变更整条请求链路都会闪断。worker 数量也并不是越大越好FastAPI 是异步服务CPU 密集型任务会把整个事件循环堵住如果你的服务里有 CPU 密集操作建议拆出来放到任务队列而不是继续加 worker。Dockerfile 里还有一个常见问题直接把整个项目COPY进镜像包括.venv、node_modules、测试文件。我更推荐用.dockerignore把不需要的文件排除掉镜像体积会小很多。启动命令保持简单FROM python:3.12-slim WORKDIR /app COPY pyproject.toml ./ COPY app ./app RUN pip install --no-cache-dir . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 2]如果在 Windows 上开发和生产部署混合要注意 uvicorn 多 worker 在 Windows 下受 multiprocessing 限制调试时先单 worker线上尽量用 Linux 容器。5.3 uvicorn 多 worker 下日志“丢失”的排查思路热搜里能刷到“uvicorn fastapi 日志丢失问题”这个坑确实很常见而且通常不是框架 bug而是日志配置和进程模型不匹配。我遇到的第一种情况在多 worker 模式下每个 worker 都往 stdout 打日志采集端只做简单聚合没有按进程分文件结果大量日志因为写入竞争看起来像“丢了”。排查方法是先确认每个 worker 是否独立 stdout再确认采集端是行拼接还是结构化采集。第二种情况更隐蔽在create_app()里调了logging.basicConfig()但 uvicorn 内部 logger 用的是logging.getLogger(uvicorn)basicConfig 的根配置并没有覆盖到它。表现就是 print 能看到logger.info看不到。建议单独建一个app/core/logging.py显式配置 uvicorn 和 app 的 logger或用 JSON 结构化日志输出这样多 worker 下的日志才能被统一采集。第三种情况是用了 gunicorn 作为进程管理器启动 uvicorn worker日志配置被加载了两次输出重复或线程安全出问题。我的做法是尽量用一种进程模型启动服务要么直接 uvicorn要么 gunicornuvicorn worker不要混用。日志结构看起来和业务无关但它决定了线上问题能不能被快速定位属于项目结构里必须提前留位子的部分。说到底FastAPI 项目结构的最佳实践不是一套固定的目录树而是一套让代码在变大的过程中仍然能够被快速定位、安全修改、轻松测试的规则。我更愿意把它看作一个会生长的结构先接受单文件阶段的粗暴再在业务明确时切换到业务域模块最后根据团队规模决定要不要上应用工厂、仓储层和独立任务模块。每一次升级都应该是业务复杂度逼着你做的而不是为了在架构图上多画几个方块。
返回列表