# Python 后端编码规范

> **版本**: v1.0 | **修订日期**: 2026-07-15
> **适用范围**: 基于 Python + FastAPI + SQLAlchemy + SQLite 的后端项目

---

## 一、技术栈

| 组件       | 版本    | 用途                 |
|------------|---------|----------------------|
| Python     | 3.11+   | 运行时               |
| FastAPI    | 0.115+  | Web 框架             |
| SQLAlchemy | 2.0+    | ORM / 数据库抽象     |
| Alembic    | 1.13+   | 数据库迁移           |
| Pydantic   | 2.0+    | 数据验证与序列化     |
| Celery     | 5.4+    | 异步任务队列         |
| Redis      | 7.0+    | 缓存 / 消息代理 / 锁 |
| Uvicorn    | 0.30+   | ASGI 服务器          |
| pytest     | 8.0+    | 单元测试             |
| httpx      | 0.27+   | HTTP 客户端 / 测试   |
| Ruff       | 0.5+    | Linter + Formatter   |

---

## 二、项目结构

### 2.1 标准目录布局

```
project-root/
├── app/
│   ├── main.py                   # FastAPI 入口
│   ├── core/                     # config, database, security, exceptions
│   ├── api/v1/endpoints/         # 按业务域划分路由
│   ├── models/                   # SQLAlchemy 模型
│   ├── schemas/                  # Pydantic Schema
│   ├── services/                 # 业务逻辑层
│   ├── repositories/             # 数据访问层（可选）
│   ├── tasks/                    # Celery 异步任务
│   └── utils/                    # 工具函数
├── alembic/                      # 数据库迁移
├── tests/                        # 测试
└── pyproject.toml
```

### 2.2 分层职责

| 层            | 职责                             | 禁止事项               |
|---------------|----------------------------------|------------------------|
| API 路由层    | 请求解析、参数校验、调用 Service | 直接操作数据库         |
| Service 层    | 业务逻辑编排、事务管理           | 直接构造 HTTP Response |
| Repository 层 | 数据库 CRUD 封装                 | 包含业务判断逻辑       |
| Model 层      | SQLAlchemy ORM 定义              | 包含业务方法           |
| Schema 层     | Pydantic 请求/响应               | 包含数据库操作         |

---

## 三、命名规范

### 3.1 类命名（PascalCase）

| 类型            | 后缀         | 示例                     |
|-----------------|--------------|--------------------------|
| 模型            | 无           | Warehouse, MaterialLot   |
| 创建请求        | Create       | WarehouseCreate          |
| 更新请求        | Update       | WarehouseUpdate          |
| 响应            | Response     | WarehouseResponse        |
| Service         | Service      | WarehouseService         |
| Repository      | Repository   | WarehouseRepository      |
| 异常            | Error        | WarehouseNotFoundError   |

### 3.2 函数命名（snake_case）

| 操作     | 模式                      | 示例                          |
|----------|---------------------------|-------------------------------|
| 创建     | create_{entity}           | create_warehouse              |
| 更新     | update_{entity}           | update_warehouse              |
| 删除     | delete_{entity}           | delete_warehouse              |
| 获取     | get_{entity}              | get_warehouse                 |
| 分页     | get_{entity}_list         | get_warehouse_list            |
| 校验     | validate_{condition}      | validate_warehouse_exists     |
| 异步任务 | {action}_{entity}_task    | sync_inventory_task           |

### 3.3 变量与常量

- 变量: snake_case，布尔用 is_/has_/can_ 前缀
- 常量: 全大写 MAX_PAGE_SIZE, DEFAULT_TIMEOUT
- 禁止单字母变量和魔法值

---

## 四、API 接口规范

### 4.1 RESTful 路由

```python
from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix="/warehouses", tags=["仓库管理"])

@router.post("", status_code=201, response_model=WarehouseResponse)
async def create_warehouse(
    data: WarehouseCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """创建仓库"""
    return await warehouse_service.create(db, data, current_user)

@router.get("/{warehouse_id}", response_model=WarehouseResponse)
async def get_warehouse(warehouse_id: int, db: AsyncSession = Depends(get_db)):
    """获取仓库详情"""
    return await warehouse_service.get_by_id(db, warehouse_id)

@router.get("", response_model=PageResponse[WarehouseListItem])
async def list_warehouses(
    params: WarehouseListParams = Depends(),
    db: AsyncSession = Depends(get_db),
):
    """分页查询仓库列表"""
    return await warehouse_service.get_list(db, params)

@router.put("/{warehouse_id}", response_model=WarehouseResponse)
async def update_warehouse(
    warehouse_id: int, data: WarehouseUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """更新仓库"""
    return await warehouse_service.update(db, warehouse_id, data, current_user)

@router.delete("/{warehouse_id}", status_code=204)
async def delete_warehouse(warehouse_id: int, db: AsyncSession = Depends(get_db)):
    """删除仓库"""
    await warehouse_service.delete(db, warehouse_id)
```

### 4.2 统一响应格式

```python
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    code: int = 0
    message: str = "success"
    data: Optional[T] = None

class PageResponse(BaseModel, Generic[T]):
    code: int = 0
    message: str = "success"
    data: PageData[T]

class PageData(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    page_size: int
```

### 4.3 错误码体系

```python
from enum import IntEnum

class ErrorCode(IntEnum):
    SUCCESS = 0
    BAD_REQUEST = 1000
    UNAUTHORIZED = 1001
    FORBIDDEN = 1003
    NOT_FOUND = 1004
    VALIDATION_ERROR = 1010
    INTERNAL_ERROR = 1999
    # 业务错误 2xxx-9xxx 按模块划分
    WAREHOUSE_NOT_FOUND = 2001
    WAREHOUSE_CODE_DUPLICATE = 2002
    MATERIAL_LOT_NOT_FOUND = 3001
```

---

## 五、Service 层规范

```python
class WarehouseService:
    """仓库管理服务"""

    async def create(self, db: AsyncSession, data: WarehouseCreate, current_user: User) -> Warehouse:
        await self._validate_code_unique(db, data.code)
        warehouse = Warehouse(**data.model_dump(), created_by=current_user.id)
        db.add(warehouse)
        await db.commit()
        await db.refresh(warehouse)
        return warehouse

    async def get_by_id(self, db: AsyncSession, warehouse_id: int) -> Warehouse:
        warehouse = await db.get(Warehouse, warehouse_id)
        if not warehouse or warehouse.is_deleted:
            raise BusinessError(ErrorCode.WAREHOUSE_NOT_FOUND)
        return warehouse

    async def update(self, db: AsyncSession, warehouse_id: int, data: WarehouseUpdate, current_user: User) -> Warehouse:
        warehouse = await self.get_by_id(db, warehouse_id)
        for field, value in data.model_dump(exclude_unset=True).items():
            setattr(warehouse, field, value)
        warehouse.updated_by = current_user.id
        await db.commit()
        await db.refresh(warehouse)
        return warehouse

    async def delete(self, db: AsyncSession, warehouse_id: int, current_user: User) -> None:
        warehouse = await self.get_by_id(db, warehouse_id)
        warehouse.is_deleted = True
        warehouse.updated_by = current_user.id
        await db.commit()

    async def _validate_code_unique(self, db: AsyncSession, code: str, exclude_id: int | None = None):
        stmt = select(Warehouse).where(Warehouse.code == code, Warehouse.is_deleted == False)
        if exclude_id:
            stmt = stmt.where(Warehouse.id != exclude_id)
        result = await db.execute(stmt)
        if result.scalar_one_or_none():
            raise BusinessError(ErrorCode.WAREHOUSE_CODE_DUPLICATE)

warehouse_service = WarehouseService()  # 模块级单例
```

**编写规则**:
- Service 方法负责 commit，Repository 层不提交
- 业务异常用 BusinessError，禁止裸 raise Exception
- 用 Pydantic Schema 接收参数，禁止裸 dict
- 返回 Model 实例，由 API 层转 Schema
- 校验用 _validate_ 前缀私有方法
- 模块级单例实例化

---

## 六、Pydantic Schema 规范

```python
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime

class WarehouseBase(BaseModel):
    code: str = Field(..., min_length=1, max_length=50, description="仓库编码")
    name: str = Field(..., min_length=1, max_length=200, description="仓库名称")
    location: str | None = Field(None, max_length=500, description="仓库位置")

class WarehouseCreate(WarehouseBase):
    pass

class WarehouseUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=200)
    location: str | None = Field(None, max_length=500)

class WarehouseResponse(WarehouseBase):
    model_config = ConfigDict(from_attributes=True)
    id: int
    is_active: bool
    created_at: datetime
    updated_at: datetime

class WarehouseListItem(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    code: str
    name: str
    is_active: bool
```

**规则**: Base 定义公共字段 | Create/Update/Response 继承复用 | Response 设 from_attributes=True | Update 全 Optional + exclude_unset

---

## 七、异常处理

```python
class BusinessError(Exception):
    def __init__(self, code: ErrorCode, message: str | None = None):
        self.code = code
        self.message = message or ERROR_MESSAGES.get(code, "未知错误")

class NotFoundError(BusinessError):
    def __init__(self, entity: str = "资源"):
        super().__init__(ErrorCode.NOT_FOUND, f"{entity}不存在")

# 全局异常处理
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(status_code=200, content={"code": int(exc.code), "message": exc.message, "data": None})

@app.exception_handler(Exception)
async def global_error_handler(request: Request, exc: Exception):
    logger.exception("Unhandled exception", exc_info=exc)
    return JSONResponse(status_code=500, content={"code": 1999, "message": "服务器内部错误", "data": None})
```

---

## 八、配置管理

```python
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_config = {"env_file": ".env"}
    APP_NAME: str = "MyApp"
    DEBUG: bool = False
    DATABASE_URL: str = "sqlite+aiosqlite:///./data/app.db"
    REDIS_URL: str = "redis://localhost:6379/0"
    CELERY_BROKER_URL: str = "redis://localhost:6379/1"
    CELERY_RESULT_BACKEND: str = "redis://localhost:6379/2"
    SECRET_KEY: str = "change-me-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

settings = Settings()
```

---

## 九、代码质量

- Ruff 作为 linter + formatter（py311, line-width=120）
- 所有公共函数必须有类型注解 + Docstring
- I/O 操作必须 async/await
- 导入排序: 标准库 → 第三方 → 本地（Ruff isort）
- 最大函数长度: 50 行
- 最大参数个数: 5 个（超出用 Pydantic Model）
- 禁止魔法值，常量全大写

---

## 十、测试规范

命名: `test_{方法名}_{场景}_{期望结果}`

```python
# tests/conftest.py
@pytest.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async_session = async_sessionmaker(engine, expire_on_commit=False)
    async with async_session() as session:
        yield session
    await engine.dispose()

@pytest.fixture
async def client(db_session: AsyncSession):
    app.dependency_overrides[get_db] = lambda: db_session
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()
```

---

## 十一、安全规范

| 规则       | 说明                                         |
|------------|----------------------------------------------|
| 密码存储   | bcrypt / argon2，禁止明文或 MD5              |
| Token      | JWT + 合理过期时间                           |
| SQL 注入   | SQLAlchemy ORM 参数化，禁止字符串拼接        |
| 敏感信息   | 环境变量注入，禁止硬编码                     |
| CORS       | 生产环境明确配置允许的 Origin                |
| 输入校验   | 全部通过 Pydantic Schema 校验                |