后端开发规范 (Backend Development)

定位：Python 后端服务的开发标准，确保代码可维护、高性能、安全。

技术选型

层级	技术栈	版本要求	选型理由
Web 框架	FastAPI	≥ 0.100	异步优先、自动文档、类型安全
ORM	SQLAlchemy	2.0+	异步支持、成熟生态、灵活查询
数据库	MySQL	8.0+	事务支持、索引优化、成熟稳定
异步驱动	asyncmy	最新	MySQL 异步驱动，配合 SQLAlchemy
数据校验	Pydantic	2.0+	类型推断、自动文档、性能优秀
迁移工具	Alembic	最新	数据库版本控制
任务队列	Celery	可选	异步任务、定时调度

项目结构

src/
├── core/               # 核心基础设施
│   ├── config.py       # 配置管理（统一入口）
│   ├── database.py     # 数据库连接池
│   ├── models.py       # ORM 模型定义
│   └── dependencies.py # FastAPI 依赖注入
├── service/            # 业务逻辑层（强制路由）
│   ├── insurance/      # 保险业务
│   ├── content/        # 内容业务
│   └── crawl/          # 爬虫业务
├── api/                # 路由层（可选，大型项目）
│   └── v1/
│       └── user.py     # API 端点
├── schemas/            # Pydantic 模型
│   └── user.py         # 请求/响应模型
├── LLM/                # LLM 集成
│   ├── factory.py      # 工厂模式入口
│   └── *_provider.py   # 各厂商适配器
├── utils/              # 工具函数
└── temp/               # 临时脚本（用完即删）

Service Layer 强制路由

规则	说明
优先检查	任何数据库操作必须先查 `src/service/` 是否有对应 Service
禁止越级	严禁在业务脚本中直接写 SQL 或调用 `db.execute()`
即时重构	缺少 Service 时，必须先在 `src/service/` 新建，再调用

# ✅ 正确：通过 Service 层
from src.service.insurance import ClauseService
result = await ClauseService.get_by_id(clause_id)

# ❌ 错误：直接操作数据库
user = await db.execute(select(User).where(User.id == 1))

数据库配置

连接池参数

参数	建议值	说明
`pool_size`	CPU 核心 × 2	并发控制
`max_overflow`	0	防止突发连接打爆数据库
`pool_pre_ping`	True	必须，网络闪断后自动重连
`echo`	False	生产环境关闭

异步会话管理

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with session_factory() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

ORM 模型规范

from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

API 规范

路由定义

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

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

@router.get("/{uid}", response_model=UserRead)
async def read_user(
    uid: int,
    session: AsyncSession = Depends(get_session)
):
    """获取用户详情"""
    return await UserService.get_by_id(session, uid)

响应格式

# 统一响应结构
class Response(BaseModel):
    code: int = 0
    message: str = "success"
    data: T | None = None

错误处理

HTTP 状态码	场景
400	参数校验失败
401	未授权
403	权限不足
404	资源不存在
422	Pydantic 校验失败
500	服务器内部错误

配置协议

规则	说明
凭证管理	所有敏感信息存储于 `.env`，严禁硬编码
统一入口	`from src.core.config import Config`，禁止各模块自行 `load_dotenv()`
安全防线	`.env` 必须加入 `.gitignore`

# src/core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DB_URL: str
    ACTIVE_LLM: str = "deepseek"

    class Config:
        env_file = ".env"

Config = Settings()

常见错误速查

异常	原因	解法
`greenlet_spawn has not been called`	用了同步引擎	使用 `create_async_engine`
`DetachedInstanceError`	会话关闭后访问属性	设置 `expire_on_commit=False`
`InterfaceError: connection already closed`	协程间复用 Session	一个请求一个 Session
`ImportError: asyncmy`	MySQL 驱动未装	`pip install asyncmy`

安全规范

规则	说明
SQL 注入防护	使用 ORM 参数化查询，禁止字符串拼接
敏感数据脱敏	日志中禁止输出密码、Token 等
输入校验	所有外部输入必须经过 Pydantic 校验
CORS 配置	生产环境限制允许的域名

测试规范

# 使用 pytest-asyncio
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/users", json={"name": "test"})
    assert response.status_code == 200

沉淀协议

优先级	触发条件	行动
P0	防御性规则、显式指令	立即沉淀到 L2_sop
P1	流程优化	提案确认后沉淀
P2	一次性操作	过滤忽略