FastAPI 项目结构设计
以下是一个基于 FastAPI 构建的后端项目结构,其遵循分层设计的理念。尽管 FastAPI 官方并未强制规定必须遵循某种特定的项目结构,给予了开发者极大的自由度,但在实际开发过程中,强烈推荐维护一个清晰且合理的目录架构。通过将项目功能进行合理拆分,可以有效降低各模块之间的耦合度,从而提升代码的可维护性、可扩展性以及可读性。
项目结构总览
my_fastapi_project/
├── app/ # 应用程序核心代码
│ ├── main.py # 应用程序入口
│ ├── api/ # API 相关模块
│ │ ├── v1/ # API 版本 1
│ │ │ ├── routes/ # 路由定义
│ │ │ └── api_v1.py # v1 API 的路由挂载
│ │ └── api_router.py # API 路由的总入口
│ ├── middleware/ # 中间件
│ ├── exceptions/ # 自定义异常
│ ├── schemas/ # 数据模型(Pydantic Schemas)
│ ├── services/ # 业务逻辑层
│ ├── repositories/ # 数据访问层
│ ├── models/ # 数据库模型(ORM)
│ ├── utils/ # 工具函数
│ └── tests/ # 测试用例
├── core/ # 核心配置和工具
│ ├── config.py # 项目配置
│ ├── deps.py # 依赖注入
│ ├── logger.py # 日志配置
│ ├── db/ # 数据库相关
│ │ ├── base.py # 数据库基类
│ │ └── session.py # 数据库会话管理
│ └── ... # 其他核心模块(按需扩展)
├── .env # 环境变量配置文件
├── README.md # 项目说明文档
└── requirements.txt # 项目依赖列表
详细说明
根目录文件
.env
- 作用: 存储环境变量,例如数据库连接字符串、API 密钥等。
- 建议: 使用
python-dotenv
库加载环境变量,避免将敏感信息硬编码到代码中。 - 示例内容:
DATABASE_URL=postgresql://user:password@localhost:5432/dbname SECRET_KEY=your-secret-key ENVIRONMENT=development
- 注意: 不应将
.env
文件提交到版本控制系统中,建议将其加入.gitignore
。
README.md
- 作用: 项目说明文档,提供项目概述、安装步骤、使用方法等。
- 建议: 包含以下部分:
- 项目简介
- 环境要求(如 Python 版本)
- 安装步骤(如
pip install -r requirements.txt
) - 启动命令(如
uvicorn app.main:app --reload
) - API 示例或测试方法
requirements.txt
- 作用: 列出项目所需的 Python 依赖包。
- 建议: 使用
pip freeze > requirements.txt
生成,确保依赖版本一致。 - 示例内容:
fastapi==0.95.0 uvicorn==0.21.1 sqlalchemy==2.0.0 pydantic==1.10.0 python-dotenv==1.0.0
APP 目录 - 应用程序核心代码
main.py
- 作用: FastAPI 应用程序的入口文件,负责初始化应用并挂载路由、中间件等。
- 实现建议:
- 创建
FastAPI
实例。 - 从
api_router.py
导入并挂载 API 路由。 - 配置中间件和异常处理。
- 创建
- 示例代码:
from fastapi import FastAPI from app.api.api_router import api_router from app.middleware import add_middleware app = FastAPI(title="My FastAPI Project", version="1.0.0") app.include_router(api_router) add_middleware(app) @app.get("/") async def root(): return {"message": "Welcome to My FastAPI Project"}
api/
目录- 作用: 包含所有 API 相关的代码,按版本组织。
api_router.py
- 作用: API 路由的总入口,负责聚合所有版本的路由。
- 实现建议: 使用
APIRouter
定义路由并挂载子版本路由。 - 示例代码:
from fastapi import APIRouter from app.api.v1.api_v1 import api_v1_router api_router = APIRouter(prefix="/api") api_router.include_router(api_v1_router)
v1/
目录api_v1.py
- 作用: v1 版本的路由入口,挂载具体路由。
- 示例代码:
from fastapi import APIRouter from app.api.v1.routes import user, item api_v1_router = APIRouter(prefix="/v1") api_v1_router.include_router(user.router) api_v1_router.include_router(item.router)
routes/
目录- 作用: 存放具体路由定义,按功能模块划分(如
user.py
,item.py
)。 - 实现建议: 每个文件定义一个
APIRouter
实例,包含相关端点。 - 示例代码 (
user.py
):from fastapi import APIRouter, Depends from app.schemas.user import UserCreate, UserResponse from app.services.user_service import create_user router = APIRouter(prefix="/users", tags=["users"]) @router.post("/", response_model=UserResponse) async def create_new_user(user: UserCreate): return await create_user(user)
- 作用: 存放具体路由定义,按功能模块划分(如
middleware/
目录- 作用: 存放自定义中间件,例如日志记录、认证、CORS 等。
- 实现建议: 定义中间件函数并在
main.py
中注册。 - 示例代码 (
middleware/log.py
):from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware import time class LogMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time print(f"{request.method} {request.url} - {process_time}s") return response def add_middleware(app): app.add_middleware(LogMiddleware)
exceptions/
目录- 作用: 定义自定义异常类,便于统一错误处理。
- 示例代码 (
exceptions/custom.py
):from fastapi import HTTPException class NotFoundException(HTTPException): def __init__(self, detail: str = "Resource not found"): super().__init__(status_code=404, detail=detail)
schemas/
目录- 作用: 使用 Pydantic 定义请求和响应的数据模型。
- 示例代码 (
schemas/user.py
):from pydantic import BaseModel class UserCreate(BaseModel): username: str email: str class UserResponse(BaseModel): id: int username: str email: str class Config: orm_mode = True # 兼容 ORM 模型
services/
目录- 作用: 业务逻辑层,处理核心功能,调用
repositories
层访问数据。 - 示例代码 (
services/user_service.py
):from app.schemas.user import UserCreate, UserResponse from app.repositories.user_repo import UserRepository async def create_user(user: UserCreate) -> UserResponse: repo = UserRepository() db_user = await repo.create(user.dict()) return UserResponse.from_orm(db_user)
- 作用: 业务逻辑层,处理核心功能,调用
repositories/
目录- 作用: 数据访问层,封装数据库操作。
- 示例代码 (
repositories/user_repo.py
):from app.models.user import User from core.db.session import get_db class UserRepository: def __init__(self): self.db = get_db() async def create(self, user_data: dict) -> User: user = User(**user_data) self.db.add(user) self.db.commit() self.db.refresh(user) return user
models/
目录- 作用: 定义数据库模型,通常使用 SQLAlchemy ORM。
- 示例代码 (
models/user.py
):from sqlalchemy import Column, Integer, String from core.db.base import Base class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) email = Column(String, unique=True, index=True)
utils/
目录- 作用: 存放通用工具函数,如加密、格式化等。
- 示例代码 (
utils/crypto.py
):from hashlib import sha256 def hash_password(password: str) -> str: return sha256(password.encode()).hexdigest()
tests/
目录- 作用: 存放测试用例,使用
pytest
等框架。 - 建议: 为每个模块编写单元测试和集成测试。
- 示例代码 (
tests/test_user.py
):from fastapi.testclient import TestClient from app.main import app client = TestClient(app) def test_create_user(): response = client.post("/api/v1/users/", json={"username": "test", "email": "[email protected]"}) assert response.status_code == 200 assert response.json()["username"] == "test"
- 作用: 存放测试用例,使用
CORE 目录 - 核心配置和工具
config.py
- 作用: 管理项目配置,如数据库 URL、日志级别等。
- 示例代码:
from pydantic import BaseSettings from dotenv import load_dotenv load_dotenv() class Settings(BaseSettings): DATABASE_URL: str SECRET_KEY: str ENVIRONMENT: str = "development" class Config: env_file = ".env" settings = Settings()
deps.py
- 作用: 定义依赖注入函数,如数据库会话。
- 示例代码:
from fastapi import Depends from core.db.session import SessionLocal def get_db(): db = SessionLocal() try: yield db finally: db.close()
logger.py
- 作用: 配置日志记录。
- 示例代码:
import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("my_fastapi_project")
db/
目录base.py
- 作用: 定义 SQLAlchemy 的基类。
- 示例代码:
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base()
session.py
- 作用: 管理数据库会话。
- 示例代码:
from sqlalchemy.orm import sessionmaker from core.config import settings from sqlalchemy import create_engine engine = create_engine(settings.DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
...
- 作用: 可扩展其他核心模块,如缓存配置、认证模块等。
使用建议
启动项目:
- 使用
uvicorn app.main:app --reload
在开发模式下运行。 - 生产环境建议使用 Gunicorn + Uvicorn。
- 使用
扩展性:
- 添加新 API 版本时,复制
v1/
目录并调整版本号。 - 新增功能时,按模块在
routes/
、services/
、repositories/
中扩展。
- 添加新 API 版本时,复制
测试:
- 使用
pytest
运行测试,确保覆盖率。
- 使用
部署:
- 配置
.env
文件,调整config.py
中的生产环境参数。
- 配置