python-backend-expert

Python 后端专家

适用场景

在以下情况下启用此技能：

• 创建或修改 FastAPI 路由处理器（端点）
• 定义或更新 Pydantic v2 请求/响应模式
• 编写 SQLAlchemy 2.0 异步模型、查询或关系
• 为数据访问实现仓储模式
• 编写服务层业务逻辑
• 创建或运行 Alembic 迁移
• 使用 Depends() 设置依赖注入链
• 处理路由/服务/仓储各层的错误

以下情况请勿使用此技能：

• 为后端代码编写测试（使用 pytest-patterns）
• FastAPI 框架机制——中间件、WebSocket、OpenAPI 自定义、CORS、生命周期（使用 fastapi-patterns）
• 部署或 CI/CD 流水线配置（使用 deployment-pipeline）
• API 契约设计或端点规划（使用 api-design-patterns）
• 架构决策或分层设计（使用 system-architecture）

说明

项目结构

app/
├── main.py              # FastAPI application factory
├── core/
│   ├── config.py        # pydantic-settings configuration
│   ├── database.py      # Async engine, session factory
│   └── security.py      # Password hashing, JWT utilities
├── models/              # SQLAlchemy ORM models
│   ├── __init__.py
│   ├── base.py          # Declarative base
│   └── user.py
├── schemas/             # Pydantic v2 schemas
│   ├── __init__.py
│   └── user.py
├── repositories/        # Data access layer
│   ├── __init__.py
│   └── user_repo.py
├── services/            # Business logic layer
│   ├── __init__.py
│   └── user_service.py
├── routes/              # FastAPI routers
│   ├── __init__.py
│   └── users.py
├── dependencies/        # Reusable Depends() providers
│   ├── __init__.py
│   └── auth.py
└── exceptions.py        # Domain exception classes

FastAPI 端点模式

每个端点都遵循以下结构：

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.create_user(data)
        return UserResponse.model_validate(user)
    except ConflictError as e:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.get_user(user_id)
        return UserResponse.model_validate(user)
    except NotFoundError as e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

规则：

• 路由只处理 HTTP 相关事项：状态码、HTTPException、响应格式化
• 路由调用服务，绝不直接调用仓储
• 使用 response_model 自动序列化响应并生成 OpenAPI 文档
• 使用 status.HTTP_* 常量，不要使用裸整数
• 使用 Depends() 注入会话、身份验证和服务

仓储模式

仓储封装所有数据库访问：

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

from app.models.user import User


class UserRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

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

    async def get_by_email(self, email: str) -> User | None:
        result = await self._session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def list_with_posts(
        self, *, offset: int = 0, limit: int = 20
    ) -> list[User]:
        result = await self._session.execute(
            select(User)
            .options(selectinload(User.posts))
            .offset(offset)
            .limit(limit)
        )
        return list(result.scalars().all())

    async def create(self, user: User) -> User:
        self._session.add(user)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def update(self, user: User, **kwargs: object) -> User:
        for key, value in kwargs.items():
            setattr(user, key, value)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def delete(self, user: User) -> None:
        await self._session.delete(user)
        await self._session.flush()

规则：

• 每个模型（或聚合根）对应一个仓储
• 仓储返回模型实例或 None，绝不返回 HTTP 响应
• 仓储中不包含业务逻辑
• 调用 add() 后始终执行 flush() + refresh()，以获取生成的字段（id、时间戳）
• 在异步上下文中使用 selectinload() 预加载关系
• 绝不从仓储抛出 HTTPException

服务层模式

服务层包含业务逻辑并协调仓储：

from app.exceptions import ConflictError, NotFoundError
from app.models.user import User
from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserPatch
from app.core.security import hash_password


class UserService:
    def __init__(self, session: AsyncSession) -> None:
        self.repo = UserRepository(session)

    async def create_user(self, data: UserCreate) -> User:
        # Business rule: email must be unique
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError(f"Email {data.email} already registered")

        # Business logic: hash password before storing
        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            display_name=data.display_name,
        )
        return await self.repo.create(user)

    async def get_user(self, user_id: int) -> User:
        user = await self.repo.get_by_id(user_id)
        if user is None:
            raise NotFoundError(f"User {user_id} not found")
        return user

    async def update_user(self, user_id: int, data: UserPatch) -> User:
        user = await self.get_user(user_id)
        update_fields = data.model_dump(exclude_unset=True)
        if "password" in update_fields:
            update_fields["hashed_password"] = hash_password(update_fields.pop("password"))
        return await self.repo.update(user, **update_fields)

规则：

• 服务层抛出领域异常（NotFoundError、ConflictError），绝不抛出 HTTPException
• 服务层是唯一包含业务逻辑的地方
• 服务层通过仓储访问数据，绝不运行原始查询
• 服务层通过构造函数接收 AsyncSession，并自行创建仓储实例
• 服务层在调用仓储之前验证业务规则

领域异常

定义领域异常层次结构：

class AppError(Exception):
    """Base application error."""

class NotFoundError(AppError):
    """Resource not found."""

class ConflictError(AppError):
    """Resource conflict (duplicate, version mismatch)."""

class ValidationError(AppError):
    """Business rule violation."""

class PermissionError(AppError):
    """Insufficient permissions."""

在 FastAPI 应用中注册全局异常处理器：

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

app = FastAPI()

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError) -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": str(exc), "code": "NOT_FOUND"})

@app.exception_handler(ConflictError)
async def conflict_handler(request: Request, exc: ConflictError) -> JSONResponse:
    return JSONResponse(status_code=409, content={"detail": str(exc), "code": "CONFLICT"})

这样，服务层可以在不了解 HTTP 的情况下抛出领域异常，路由也无需使用 try/except 代码块。

Pydantic v2 模式约定

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


class UserCreate(BaseModel):
    """POST request body — writable fields only, no id/timestamps."""
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    display_name: str = Field(min_length=1, max_length=100)


class UserPatch(BaseModel):
    """PATCH request body — all fields Optional."""
    email: EmailStr | None = None
    password: str | None = Field(default=None, min_length=8, max_length=128)
    display_name: str | None = Field(default=None, min_length=1, max_length=100)


class UserResponse(BaseModel):
    """Response body — all fields including id and timestamps."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    display_name: str
    is_active: bool
    created_at: datetime
    updated_at: datetime

Pydantic v2 主要模式：

• 使用 ConfigDict(from_attributes=True)，而不是 class Config: orm_mode = True
• 使用 model_validate()，而不是 from_orm()
• 使用 model_dump()，而不是 .dict()
• 使用 model_dump(exclude_unset=True) 处理 PATCH，以区分“未发送”和“设为 null”
• 使用 Field() 设置验证约束
• 使用 str | None 语法（Python 3.12+），不要使用 Optional[str]

异步会话管理

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        async with session.begin():
            yield session

规则：

• expire_on_commit=False 可防止提交后出现分离实例错误
• session.begin() 上下文管理器在成功时自动提交，在发生异常时回滚
• 每个请求使用一个会话，通过 Depends(get_async_session) 注入
• 绝不在并发任务间共享会话
• 对于后台任务，创建新会话，绝不复用请求会话

SQLAlchemy 2.0 模型模式

from datetime import datetime
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    display_name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    # Relationships — ALWAYS use selectin or joined for async
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", lazy="selectin"
    )

规则：

• 使用 Mapped[type] 注解（SQLAlchemy 2.0 风格）
• 使用 mapped_column()，而不是 Column()
• 在关系上设置 lazy="selectin"，以确保兼容异步操作
• 对数据库生成的默认值使用 server_default
• 始终包含 created_at 和 updated_at 时间戳

Alembic 迁移工作流

# Generate migration from model changes
alembic revision --autogenerate -m "add_users_table"

# Review the generated migration file before applying

# Apply migration
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history

迁移命名约定：

# alembic/env.py
naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

规则：

• 应用自动生成的迁移前，务必先检查
• 每个迁移都必须有可用的 downgrade() 函数
• 每项逻辑上的架构变更对应一个迁移
• 测试升级和降级
• 使用描述性迁移消息："add_users_table"、"add_email_index_to_users"

依赖注入模式

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_async_session
from app.services.user_service import UserService


async def get_user_service(
    session: AsyncSession = Depends(get_async_session),
) -> UserService:
    return UserService(session)


# Chain dependencies for auth
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    user_id = decode_token(token)
    service = UserService(session)
    return await service.get_user(user_id)


async def require_admin(
    user: User = Depends(get_current_user),
) -> User:
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin required")
    return user

示例

完整请求流程

对 POST /users 的请求会经过所有层：

1. 路由 接收 UserCreate（Pydantic 会验证请求体）
2. 路由 通过 Depends() 调用 UserService.create_user(data)
3. 服务 通过 UserRepository.get_by_email() 检查业务规则（电子邮件唯一性）
4. 服务 对密码进行哈希处理，并创建 User 模型实例
5. 服务 调用 UserRepository.create(user) 进行持久化
6. 仓储 将对象添加到会话，执行刷新，并刷新对象以获取生成的字段
7. 路由 通过 model_validate() 将 ORM 模型转换为 UserResponse

如果电子邮件重复，服务会引发 ConflictError，全局异常处理程序会返回 409 Conflict。路由中无需使用 try/except。

边缘情况

• 分离实例错误： 在 session.add() 后，务必调用 flush() + refresh()。在会话工厂上设置 expire_on_commit=False。

• 后台任务中的异步会话： 切勿复用请求会话。创建新会话：

  async def background_job():
      async with async_session_factory() as session:
          async with session.begin():
              # do work
  

• N+1 查询： 对于将要访问的关系，在仓储查询中使用 selectinload()。将模型关系的默认值设为 lazy="selectin"。

• 批量操作： 批量插入时，使用 session.execute(insert(User).values(list_of_dicts))，而不是逐条添加。

• 跨多个服务的事务： 将同一个会话传递给所有服务。会话的 begin() 上下文管理器会处理事务边界。

• Pydantic v2 计算字段： 在响应模式中使用 @computed_field 表示派生值。参见 references/pydantic-v2-migration.md。

有关高级查询优化模式，请参见 references/sqlalchemy-patterns.md。