fastapi-coder

FastAPI 编码器

核心原则

原则	应用方式
异步优先	全面使用 async/await，仅在必要时使用同步代码
类型安全	所有请求/响应数据都使用 Pydantic 模型
依赖注入	使用 Depends() 处理共享逻辑，而非全局状态
OpenAPI 驱动	自动生成 Schema；保持整洁
关注点分离	路由 → 服务 → 仓库

项目结构

app/
├── main.py              # FastAPI app initialization
├── api/
│   ├── __init__.py
│   ├── deps.py          # Shared dependencies
│   └── routes/          # Route handlers by domain
│       ├── users.py
│       └── items.py
├── core/
│   ├── config.py        # Settings via pydantic-settings
│   ├── security.py      # Auth utilities
│   └── exceptions.py    # Custom exceptions
├── models/              # Pydantic schemas
│   ├── user.py
│   └── item.py
├── services/            # Business logic
│   └── user_service.py
├── repositories/        # Data access
│   └── user_repo.py
└── tests/
    ├── conftest.py      # Shared fixtures
    └── test_users.py

核心模式

路由处理程序

from fastapi import APIRouter, Depends, HTTPException, status
from app.models.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.api.deps import get_user_service

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    """Create a new user."""
    return await service.create(user_in)

Pydantic 模型

from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}

依赖项

from typing import Annotated
from fastapi import Depends, Header, HTTPException
from app.core.security import verify_token

async def get_current_user(
    authorization: Annotated[str, Header()],
) -> User:
    token = authorization.removeprefix("Bearer ")
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

服务层

from app.repositories.user_repo import UserRepository
from app.models.user import UserCreate, UserResponse

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create(self, user_in: UserCreate) -> UserResponse:
        # Business logic here
        existing = await self.repo.get_by_email(user_in.email)
        if existing:
            raise ValueError("Email already registered")
        return await self.repo.create(user_in)

异常处理

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

后台任务

from fastapi import BackgroundTasks

async def send_welcome_email(email: str):
    # Async email sending
    ...

@router.post("/users/")
async def create_user(
    user_in: UserCreate,
    background_tasks: BackgroundTasks,
):
    user = await create_user_in_db(user_in)
    background_tasks.add_task(send_welcome_email, user.email)
    return user

数据库集成

SQLAlchemy 2.0 异步

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("postgresql+asyncpg://...", echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

仓库模式

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

身份验证模式

JWT 身份验证

from datetime import datetime, timedelta
from jose import jwt, JWTError
from app.core.config import settings

def create_access_token(data: dict) -> str:
    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({**data, "exp": expire}, settings.SECRET_KEY, algorithm="HS256")

async def verify_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None

测试模式

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={
        "email": "[email protected]",
        "name": "Test User",
        "password": "securepass123",
    })
    assert response.status_code == 201
    assert response.json()["email"] == "[email protected]"

质量检查清单

• 所有路由都设置了 response_model 和 status_code
• 所有请求/响应数据都使用 Pydantic 模型
• 使用依赖项处理共享逻辑（身份验证、数据库、服务）
• 服务层将业务逻辑与路由分离
• 使用仓库模式进行数据访问
• 自定义异常并设置适当的处理程序
• 异步数据库操作
• 使用后台任务执行非阻塞操作
• 使用 httpx AsyncClient 编写全面的测试
• 全面添加类型提示