1. 项目结构规范
myproject/
├── app/
│ ├── core/ # 核心配置
│ │ ├── config.py # 环境配置
│ │ └── security.py # 安全配置
│ ├── routers/ # 路由模块
│ │ └── users.py # 用户路由
│ ├── models/ # 数据库模型
│ │ └── user.py # 用户模型
│ ├── schemas/ # Pydantic模型
│ │ └── user.py # 用户Schema
│ ├── services/ # 业务逻辑
│ │ └── user.py # 用户服务
│ ├── dependencies.py # 依赖注入
│ └── db/ # 数据库配置
├── tests/ # 测试用例
│ └── test_users.py # 用户测试
├── requirements.txt # 依赖文件
└── main.py # 入口文件
2. 核心代码示例
配置管理 (app/core/config.py)
from pydantic import BaseSettingsclass Settings(BaseSettings):API_V1_STR: str = "/api/v1"DB_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/db"JWT_SECRET: str = "secret-key"class Config:env_file = ".env"settings = Settings()
路由示例 (app/routers/users.py)
from fastapi import APIRouter, Depends, HTTPException
from app.services.user import UserService
from app.schemas.user import UserCreate, UserResponse
from app.dependencies import get_dbrouter = APIRouter(prefix="/users", tags=["users"])@router.post("/", response_model=UserResponse)
async def create_user(user: UserCreate,service: UserService = Depends(UserService)
):try:return await service.create_user(user)except Exception as e:raise HTTPException(400, str(e))
服务层 (app/services/user.py)
from app.models.user import User
from app.schemas.user import UserCreate, UserResponse
from app.core.security import get_password_hashclass UserService:def __init__(self, db):self.db = dbasync def create_user(self, user_create: UserCreate):hashed_password = get_password_hash(user_create.password)user = User(email=user_create.email,hashed_password=hashed_password)self.db.add(user)await self.db.commit()return UserResponse(**user.dict())
数据库模型 (app/models/user.py)
from sqlalchemy import Column, Integer, String
from app.db.base import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True)email = Column(String, unique=True, index=True)hashed_password = Column(String)
Schema验证 (app/schemas/user.py)
from pydantic import BaseModel, EmailStrclass UserBase(BaseModel):email: EmailStrclass UserCreate(UserBase):password: strclass UserResponse(UserBase):id: intclass Config:orm_mode = True
3. 关键规范说明
-
分层架构：
- 路由层：只处理HTTP协议和参数验证
- 服务层：处理业务逻辑
- 数据访问层：处理数据库操作
-
依赖注入：
# app/dependencies.py
async def get_db():async_session = sessionmaker(expire_on_commit=False,class_=AsyncSession)async with async_session() as session:yield session
- 安全规范：
# app/core/security.py
from passlib.context import CryptContextpwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")def verify_password(plain_password, hashed_password):return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password):return pwd_context.hash(password)
- 异常处理：
# app/main.py
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationErrorapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return JSONResponse(status_code=400,content={"detail": exc.errors(), "body": exc.body},)
- 测试规范：
# tests/test_users.py
from fastapi.testclient import TestClientdef test_create_user():client = TestClient(app)response = client.post("/users/", json={"email": "test@example.com","password": "string"})assert response.status_code == 200assert "id" in response.json()
4. 最佳实践建议
- 使用异步数据库驱动（如asyncpg）
- 为每个路由添加OpenAPI标签
- 强制类型注解（mypy兼容）
- 使用alembic进行数据库迁移
- 统一响应格式：
class BaseResponse(BaseModel):code: int = 200message: str = "success"data: Any = None
- 接口版本管理：
router = APIRouter(prefix="/v1/users")
- 日志记录规范：
import logging
logging.basicConfig(format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",level=logging.INFO
)
该方案结合了FastAPI官方推荐实践和企业级开发经验，在保持灵活性的同时确保代码质量。建议配合以下工具链：
- 代码格式化：black + isort
- 静态检查：mypy + pylint
- 测试覆盖：pytest + coverage
- 文档生成：swagger UI + redoc
- CI/CD：GitHub Actions/GitLab CI
可根据具体业务需求扩展中间件、缓存、任务队列等模块，但需保持核心架构的稳定性。
