1. FastAPI项目结构设计原则
中大型FastAPI项目的标准结构需要遵循几个关键原则:模块化、可维护性和可扩展性。经过多个生产级项目的实践验证,我认为以下设计原则最为关键:
- 业务功能隔离:每个主要业务功能应该作为独立模块存在
- 依赖关系清晰:明确区分核心依赖和业务依赖
- 配置与代码分离:环境配置不应硬编码在业务逻辑中
- 测试友好:结构应便于单元测试和集成测试
典型的项目目录结构应该像这样:
project/ ├── app/ # 主应用包 │ ├── core/ # 核心基础设施 │ ├── modules/ # 业务模块 │ ├── shared/ # 共享代码 │ └── main.py # 应用入口 ├── tests/ # 测试代码 ├── configs/ # 配置文件 ├── scripts/ # 运维脚本 └── requirements/ # 依赖管理2. 核心目录结构详解
2.1 应用核心组件
app/core目录包含整个应用的基础设施:
core/ ├── __init__.py ├── config.py # 配置加载逻辑 ├── dependencies.py # 全局依赖项 ├── exceptions.py # 自定义异常 ├── middleware.py # 中间件 └── security.py # 认证授权相关配置管理是其中最重要的部分。我推荐使用pydantic的BaseSettings:
# core/config.py from pydantic import BaseSettings class Settings(BaseSettings): app_name: str = "My FastAPI App" database_url: str secret_key: str class Config: env_file = ".env"2.2 业务模块组织
业务模块应该按功能垂直划分,每个模块包含完整的业务逻辑:
modules/ ├── auth/ │ ├── routers.py │ ├── schemas.py │ ├── services.py │ └── models.py ├── products/ │ ├── routers.py │ └── ... └── users/ ├── routers.py └── ...这种结构的关键优势在于:
- 修改一个业务功能不会影响其他模块
- 每个模块可以独立测试
- 便于团队分工协作
3. 依赖管理与应用组装
3.1 依赖注入设计
FastAPI强大的依赖注入系统需要合理设计。我建议:
# core/dependencies.py from fastapi import Depends, HTTPException from sqlalchemy.orm import Session def get_db(): db = SessionLocal() try: yield db finally: db.close() async def get_current_user( db: Session = Depends(get_db), token: str = Depends(OAuth2PasswordBearer(tokenUrl="token")) ): # 用户认证逻辑 return user3.2 应用组装
主应用文件应该保持简洁:
# app/main.py from fastapi import FastAPI from .core.config import settings from .core.middleware import add_middleware app = FastAPI(title=settings.app_name) add_middleware(app) # 导入路由 from app.modules.auth import router as auth_router app.include_router(auth_router, prefix="/auth")4. 测试策略与配置
4.1 测试目录结构
测试应该镜像主代码结构:
tests/ ├── unit/ │ ├── test_services.py │ └── ... ├── integration/ │ ├── test_api.py │ └── ... └── conftest.py # pytest fixtures4.2 测试配置示例
# tests/conftest.py import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.core.config import settings from app.main import app @pytest.fixture def test_db(): engine = create_engine(settings.test_database_url) TestingSessionLocal = sessionmaker(autocommit=False, bind=engine) Base.metadata.create_all(bind=engine) db = TestingSessionLocal() try: yield db finally: db.close() Base.metadata.drop_all(bind=engine)5. 生产环境注意事项
5.1 部署优化建议
ASGI服务器选择:
- Uvicorn适合大多数场景
- 高并发考虑Hypercorn或Daphne
性能调优:
uvicorn app.main:app --workers 4 --limit-concurrency 100健康检查端点:
@app.get("/health") async def health_check(): return {"status": "healthy"}
5.2 监控与日志
建议集成:
- Prometheus指标
- Sentry错误跟踪
- 结构化日志(JSON格式)
# core/logging.py import logging from pythonjsonlogger import jsonlogger def setup_logging(): logger = logging.getLogger() handler = logging.StreamHandler() formatter = jsonlogger.JsonFormatter() handler.setFormatter(formatter) logger.addHandler(handler)6. 常见问题解决方案
6.1 循环导入问题
当模块间需要相互引用时,可以采用:
- 将共享模型移到
shared/目录 - 使用字符串类型的类型提示
def get_user_service() -> 'UserService': ...
6.2 数据库会话管理
常见陷阱及解决方案:
# 错误做法:在路由中直接处理会话 @app.get("/items") async def read_items(db: Session = Depends(get_db)): # 业务逻辑与会话管理混在一起 pass # 正确做法:使用服务层 class ItemService: def __init__(self, db: Session): self.db = db def get_items(self): return self.db.query(Item).all()7. 项目演进建议
随着项目规模扩大,可以考虑:
- 引入领域驱动设计:更清晰地划分限界上下文
- 使用Celery:处理后台任务
- API拆分:当单体过大时转为微服务
# 渐进式演进示例 @app.on_event("startup") async def startup_event(): # 初始化后台任务 from .tasks import celery_app celery_app.conf.update(app.config)