# 数据库设计与使用规范

> **版本**: v1.0 | **修订日期**: 2026-07-15
> **技术栈**: SQLAlchemy 2.0+ / Alembic / SQLite (aiosqlite)

---

## 一、SQLAlchemy 基类

### 1.1 声明式基类

```python
# app/models/base.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String, Boolean, DateTime, func
from datetime import datetime

class Base(DeclarativeBase):
    """SQLAlchemy 声明式基类"""
    pass

class TimestampMixin:
    """时间戳 Mixin — 自动记录创建/更新时间"""
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(), nullable=False
    )

class SoftDeleteMixin:
    """软删除 Mixin"""
    is_deleted: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

class AuditMixin:
    """审计 Mixin — 记录操作人"""
    created_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
    updated_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
```

### 1.2 模型定义示例

```python
# app/models/warehouse.py
from sqlalchemy import String, Integer, Text, Index
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, SoftDeleteMixin, AuditMixin

class Warehouse(Base, TimestampMixin, SoftDeleteMixin, AuditMixin):
    """仓库模型"""
    __tablename__ = "warehouses"
    __table_args__ = (
        Index("ix_warehouse_code", "code", unique=True),
        Index("ix_warehouse_active", "is_active"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    code: Mapped[str] = mapped_column(String(50), nullable=False, comment="仓库编码")
    name: Mapped[str] = mapped_column(String(200), nullable=False, comment="仓库名称")
    location: Mapped[str | None] = mapped_column(String(500), comment="位置")
    is_active: Mapped[bool] = mapped_column(default=True, comment="是否启用")

    # relationships
    material_lots: Mapped[list["MaterialLot"]] = relationship(back_populates="warehouse")
```

---

## 二、SQLite 特有注意事项

| 事项           | 规范                                                       |
|----------------|------------------------------------------------------------|
| WAL 模式       | 生产环境必须启用 `PRAGMA journal_mode=WAL`                 |
| 外键约束       | 默认关闭，必须执行 `PRAGMA foreign_keys=ON`               |
| 类型亲和性     | 严格按模型定义列类型，避免隐式转换                         |
| 并发写入       | SQLite 不支持并发写，Celery worker 必须 `--concurrency=1` |
| 文件权限       | 数据库文件目录需要读写权限                                 |
| 备份           | 定期 VACUUM + 文件拷贝                                    |

### 引擎配置

```python
# app/core/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import text
from app.core.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=settings.DEBUG,
    connect_args={"check_same_thread": False},
)

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

async def init_db():
    """初始化数据库（启用 WAL 模式）"""
    async with engine.connect() as conn:
        await conn.execute(text("PRAGMA journal_mode=WAL"))
        await conn.execute(text("PRAGMA foreign_keys=ON"))
        await conn.commit()

async def get_db():
    """数据库会话依赖注入"""
    async with async_session() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
```

---

## 三、Alembic 迁移规范

### 3.1 初始化

```bash
alembic init alembic
# 修改 alembic.ini 的 sqlalchemy.url
# 修改 alembic/env.py 导入 Base.metadata
```

### 3.2 迁移命名

```bash
alembic revision --autogenerate -m "add_warehouse_table"
alembic revision --autogenerate -m "add_index_warehouse_code"
```

### 3.3 迁移规则

| 规则       | 说明                                           |
|------------|------------------------------------------------|
| 自动生成   | 始终使用 `--autogenerate`，手动 review 后提交  |
| 不可逆操作 | DROP TABLE/DROP COLUMN 前必须确认数据安全     |
| 数据迁移   | 数据迁移和结构迁移分开为两个 revision          |
| 测试       | 迁移后必须运行测试验证                         |
| 提交       | 迁移文件必须纳入版本控制                       |

---

## 四、查询规范

### 4.1 使用 select() 风格

```python
from sqlalchemy import select, func

# 单条查询
stmt = select(Warehouse).where(Warehouse.id == warehouse_id)
result = await db.execute(stmt)
warehouse = result.scalar_one_or_none()

# 分页查询
stmt = (
    select(Warehouse)
    .where(Warehouse.is_deleted == False, Warehouse.is_active == True)
    .order_by(Warehouse.created_at.desc())
    .offset((page - 1) * page_size)
    .limit(page_size)
)
result = await db.execute(stmt)
items = result.scalars().all()

# 计数
count_stmt = select(func.count()).select_from(Warehouse).where(Warehouse.is_deleted == False)
total = (await db.execute(count_stmt)).scalar()
```

### 4.2 禁止事项

- 禁止使用 `session.query()` 风格（SQLAlchemy 1.x 遗留）
- 禁止字符串拼接 SQL
- 禁止在循环中执行查询（N+1 问题）
- 禁止不带 WHERE 的 UPDATE/DELETE

---

## 五、Repository 模式

```python
# app/repositories/warehouse.py
class WarehouseRepository:
    """仓库数据访问层"""

    async def get_by_id(self, db: AsyncSession, warehouse_id: int) -> Warehouse | None:
        return await db.get(Warehouse, warehouse_id)

    async def get_by_code(self, db: AsyncSession, code: str) -> Warehouse | None:
        stmt = select(Warehouse).where(Warehouse.code == code, Warehouse.is_deleted == False)
        result = await db.execute(stmt)
        return result.scalar_one_or_none()

    async def get_page(
        self, db: AsyncSession, page: int, page_size: int, filters: dict | None = None
    ) -> tuple[list[Warehouse], int]:
        base = select(Warehouse).where(Warehouse.is_deleted == False)
        if filters:
            if "name" in filters:
                base = base.where(Warehouse.name.contains(filters["name"]))
            if "is_active" in filters:
                base = base.where(Warehouse.is_active == filters["is_active"])

        count_stmt = select(func.count()).select_from(base.subquery())
        total = (await db.execute(count_stmt)).scalar()

        stmt = base.order_by(Warehouse.created_at.desc()).offset((page - 1) * page_size).limit(page_size)
        result = await db.execute(stmt)
        return result.scalars().all(), total

warehouse_repo = WarehouseRepository()
```

---

## 六、索引策略

| 场景           | 索引类型         | 示例                                      |
|----------------|------------------|-------------------------------------------|
| 唯一约束       | Unique Index     | code, email                               |
| 高频查询条件   | 普通索引         | is_active, status                         |
| 组合查询       | 复合索引         | (warehouse_id, material_type)             |
| 外键           | 自动创建         | relationship 自动生成                     |
| 软删除过滤     | 部分索引         | WHERE is_deleted = False                  |