数据库开发指南

本文说明 AutoBangumi 后端的数据层：引擎、Database 门面、仓储、会话约定和 schema 迁移。代码在 backend/src/module/database/。

概述

• 数据库是 SQLite 文件 data/data.db，ORM 是 SQLModel（Pydantic + SQLAlchemy）。
• 整个应用只用异步引擎 sqlite+aiosqlite。每个连接设置 PRAGMA journal_mode=WAL、busy_timeout=5000 和 foreign_keys=ON：多个连接可以并发写，冲突的写入会等待而不是立刻报 database is locked。
• 外键约束是生效的。删除被引用的行之前，要先处理引用它的行（例如 BangumiDatabase.delete_all 先删种子记录）。

module/database/
├── __init__.py          # Database、get_db（FastAPI 依赖）
├── engine.py            # async_engine、async_session_factory、连接 PRAGMA
├── combine.py           # Database：一个会话加全部仓储
├── migrations.py        # MIGRATIONS 表与迁移执行
├── bangumi.py、rss.py、torrent.py、movie.py、user.py、auth.py、
├── inbox.py、llm_credential.py、rename_operation.py、aria2.py、
└── plugin_kv.py、passkey.py   # 各表的仓储

Database 与会话

Database 是异步上下文管理器。它创建一个 AsyncSession，并把仓储挂在自己身上；退出 async with 块时关闭会话。

from module.database import Database

async with Database() as db:
    rules = await db.bangumi.search_all()
    feeds = await db.rss.search_all()

属性	仓储	表
db.bangumi	BangumiDatabase	bangumi
db.rss	RSSDatabase	rssitem
db.torrent	TorrentDatabase	torrent
db.movie	MovieDatabase	movie
db.user	UserDatabase	user
db.auth	AuthDatabase	auth_session、api_token
db.inbox	InboxDatabase	通知中心
db.llm_credential	LLMCredentialDatabase	LLM 订阅凭据
db.rename_operation	RenameOperationDatabase	rename_operation（重命名冲突与进度）
db.aria2	Aria2GidDatabase	aria2_gid
db.plugin_kv	PluginKVDatabase	plugin_kv（插件的 ctx.kv）

PasskeyDatabase 不挂在 Database 上，需要时用 PasskeyDatabase(db.session) 构造。

Database 还转发几个会话方法：add（同步，只是入队）、commit、rollback、refresh、close。begin_write() 在 SQLite 上执行 BEGIN IMMEDIATE，用于「先读后写且须保持不变量」的场景，必须在本会话的第一次访问之前调用。

会话约定

每次操作一个会话。 不要把会话或 Database 存到比一次请求或一轮循环活得更久的对象上。AppContext 不持有会话。

• 路由用依赖注入：

  from fastapi import Depends
  from module.database import Database, get_db
  
  @router.get("/example")
  async def example(db: Database = Depends(get_db)):
      return await db.bangumi.search_all()

• 定时循环与服务在每次执行时自己开：async with Database() as db:。

• 服务在构造函数里接收依赖：RSSEngine(db)、TorrentManager(db)。服务内部用 self.db.<仓储>；只需要一个仓储的调用方直接用 db.<仓储>。

• 引擎设置了 expire_on_commit=False，提交后对象的属性仍可读，但对象不再与新会话关联。需要修改时，在新的会话里重新查询。

仓储

仓储接收一个 AsyncSession，方法都是 async def，大多在方法内部提交。新增查询时：

• 放进对应表的仓储，不要在路由或服务里直接写 SQL。
• 批量查询用 in_ 一次取回，不要在循环里逐条查。
• 软删除：bangumi 和 movie 有 deleted 字段。「停用」规则（disable_rule）只把 deleted 置为 True；delete_one 才真正删除行。查询活动规则时要过滤 deleted。

多下载器

4.0 支持多个下载器实例（配置 plugins.instances）。bangumi、movie、rssitem 和 torrent 都有 downloader_id 列：

• 规则与订阅上为空表示「使用默认实例」（plugins.slots.downloader）。
• 种子记录上是种子实际所在的实例。从 3.3 升级的存量种子为 default。

Schema 迁移

迁移在 migrations.py 的 MIGRATIONS 元组中，以表驱动的方式执行。数据库里的 schema_version 表记录已应用的版本。

Migration(
    26,
    "add downloader_id to bangumi, movie, rssitem and torrent",
    (...SQL 语句...),
    already_applied=...,          # 守卫：该迁移是否已生效
    guarded_statements=(...),     # 可选：逐条语句的守卫
)

启动时 AB 先执行 SQLModel.metadata.create_all，再执行 run_migrations()：

1. 读取 schema_version，跳过版本号不大于它的迁移。
2. 对每个待应用的迁移，重新检查数据库结构，再调用守卫：已生效（例如新库已由 create_all 按模型建表）时只记录版本号。
3. 否则在一个 SAVEPOINT 中执行它的语句。任何一条失败，该迁移回滚，异常向上抛出，整个迁移事务回滚，AB 启动失败。AB 不会在只迁移了一半的 schema 上继续运行。
4. 有迁移被执行时，最后按模型的默认值填充存量行中的 NULL（fill_null_with_defaults_conn）：跳过主键、default_factory 字段和默认值为 None 的可选字段。

添加迁移

1. 先修改 module/models/ 中的模型。新库由 create_all 直接得到新结构。
2. 在 MIGRATIONS 末尾追加一个 Migration，版本号为上一个加 1。CURRENT_SCHEMA_VERSION 由列表推出，不要手动修改。
3. 提供 already_applied 守卫：column_exists(表, 列)、table_exists(表)、index_exists(表, 索引)，多个条件用 all_checks(...) 组合。一个迁移改多张表、每张表可能处于不同状态时，用 guarded_statements 给每条语句单独的守卫。
4. 在 backend/src/test/ 中为迁移加测试：从旧 schema 开始，执行迁移，检查结果。

Migration(
    27,
    "add note to bangumi",
    ("ALTER TABLE bangumi ADD COLUMN note TEXT DEFAULT ''",),
    column_exists("bangumi", "note"),
)

测试

backend/src/test/conftest.py 提供内存数据库：

• db_engine：sqlite+aiosqlite:// 加 StaticPool，每个测试新建并按模型建表。
• db_session：绑定 db_engine 的 AsyncSession。

直接测试仓储时，用 Database(engine=db_engine)，或用 db_session 构造单个仓储：

from module.database import Database
from module.database.bangumi import BangumiDatabase
from test.factories import make_bangumi

async def test_add_bangumi(db_engine):
    async with Database(engine=db_engine) as db:
        assert await db.bangumi.add(make_bangumi())

async def test_search(db_session):
    repo = BangumiDatabase(db_session)
    assert await repo.search_all() == []

后端的 pytest 已开启 asyncio_mode = "auto"，异步测试函数不需要额外的装饰器。运行：cd backend && uv run pytest src/test/test_xxx.py -v。