存储与 ORM

存储与 ORM

Lingchu Bot 通过两层协作存储运行时数据：nonebot_plugin_orm 用于关系数据，基于 TOML 的文件存储用于轻量运行时配置。本页覆盖默认后端、四个支持的数据库、跨方言类型兼容层、方言专属 upsert 实现，以及与 nonebot_plugin_localstore 协作的 TOML 存储。

默认后端

默认数据库后端是通过 aiosqlite 的 SQLite，由 nonebot_plugin_orm 提供。本地开发无需显式连接 URL——ORM 插件会在 localstore 数据目录下创建 SQLite 数据库文件。

所有关系访问通过 nonebot_plugin_orm 会话进行。项目不引入自定义引擎管理；database/orm_crud/ 暴露的是类型化异步辅助，第一个位置参数为外部传入的 AsyncSession（见下文 Session 归属与 handler 注入）。

四个支持的后端

SQLALCHEMY_DATABASE_URL（由 nonebot_plugin_orm 消费）选择后端。支持四个引擎：

后端	驱动（URL scheme）	备注
SQLite	sqlite+aiosqlite://	默认；无需 URL
PostgreSQL	postgresql+psycopg:// 或 postgresql+asyncpg://	使用 on_conflict_do_update
MySQL	mysql+aiomysql://	使用 on_duplicate_key_update
MariaDB	mariadb+aiomysql://	共用 MySQL 路径；aiomysql 驱动

SQLALCHEMY_DATABASE_URL 未设置时，nonebot_plugin_orm 回退到默认 SQLite 数据库。CI 矩阵覆盖全部四个引擎及版本变体（PostgreSQL 16/18、MySQL 8.4/9.7、MariaDB 11.4/11.8）。不支持 Oracle / SQL Server。

跨方言类型

ORM 模型 MUST 使用 database/_dialect_compat.py 中的兼容类型，而非原始 String / Text / Boolean / DateTime(timezone=True)。该模块导出四个辅助：

辅助	行为
CompatBoolean	四个后端均为原生 BOOLEAN
CompatDateTimeTZ	多数后端为 DateTime(timezone=True)；MySQL / MariaDB 为 DATETIME(fsp=6)
CompatText	SQLite / PostgreSQL 为 TEXT；MySQL / MariaDB 为 LONGTEXT
compat_string(length)	四个后端均为 VARCHAR(length)

CompatDateTimeTZ 在 MySQL / MariaDB 上会发出“timezone only supported in MySQL 5.6+“警告。写入使用 datetime.now(UTC)（database/models/message.py 中的 utc_now() 辅助），因此实践中不会出现漂移。

仓库中当前所有 String 列长度 ≤ 128，因此 compat_string(length) 在所有后端上都保持 VARCHAR(N)。

四后端 upsert

database/orm_crud/_bulk.py::upsert() 是跨全部四个后端原子 upsert 的唯一入口。它按会话 dialect name 分发：

Dialect	实现	RETURNING 支持
sqlite	sqlite_insert(model).on_conflict_do_update(...)	是——使用 RETURNING
postgresql	postgresql_insert(model).on_conflict_do_update(...)	是——使用 RETURNING
mysql、mariadb	mysql_insert(model).on_duplicate_key_update(...)	否——按 conflict_fields 后续 SELECT

所有 upsert 调用必须提供 conflict_fields 或 constraint（互斥）。MySQL / MariaDB 要求 conflict_fields，因为它们用其进行后续 SELECT 取回行。

ALEMBIC_STARTUP_CHECK

ALEMBIC_STARTUP_CHECK 是 nonebot_plugin_orm 的配置键，并非 Lingchu 专属设置。设为 true 时，ORM 插件在启动时强制执行 Alembic schema 迁移检查。生产部署应设置它：

ALEMBIC_STARTUP_CHECK=true

默认为 false 以保持本地开发快速。Docker Compose 生产模板（docker-compose.yml）发布时带 ALEMBIC_STARTUP_CHECK: "true"。

Lingchu Bot 的模型包（位于 database/models/）在 __init__.py 中导入所有模型，以便 Alembic autogenerate 发现可用。非 SQLite 测试前必须运行迁移。

迁移生成工作流

nonebot_plugin_orm 包装了 Alembic，并通过 nb orm 暴露三个 CLI 命令。revision 命令默认启用 autogenerate——不存在 --autogenerate 标志。

命令	用途
nb orm revision -m "msg" --branch-label nonebot_plugin_lingchu_bot	基于模型变更生成新迁移脚本（默认开启 autogenerate）。--branch-label 必填，用于把文件放到 src/plugins/nonebot_plugin_lingchu_bot/migrations/；不带则会落到 ./migrations/versions/
nb orm check	检测 ORM 模型与数据库 schema 之间的漂移；不匹配时抛出 AutogenerateDiffsDetected
nb orm sync	仅用于开发的直接 schema 同步，不生成迁移脚本（在 ALEMBIC_STARTUP_CHECK=false 时使用）

本项目为方便使用添加了 Taskfile 别名：

Task	等价命令
task db:revision -- MSG="..."	ENVIRONMENT=dev nb orm revision -m "..." --branch-label nonebot_plugin_lingchu_bot
task db:check	ENVIRONMENT=dev nb orm check
task db:upgrade	ENVIRONMENT=dev nb orm upgrade

推荐流程

1. 修改 database/models/*.py。
2. 运行 task db:revision -- MSG="描述变更" 生成迁移脚手架。
3. 手工后处理生成的迁移以兼容多方言（见下文）。
4. 运行 task db:upgrade 本地应用，再运行 task db:check 确认无漂移。
5. 模型与迁移一起提交。

生成后的 Compat* 改写

autogenerate 会输出通用 SQLAlchemy 类型。为兼容跨方言，需手工将其替换为 database/_dialect_compat.py 中的辅助：

自动生成	替换为
sa.Boolean()	CompatBoolean
sa.DateTime(timezone=True)	CompatDateTimeTZ
sa.Text()	CompatText
sa.String(length)	compat_string(length)

对于 unique-constraint 或 index 重建，参考 migrations/cf2c06d51a17_blocklist_unique_constraint.py，在默认 op.create_index(..., mysql_length=...) 形态不同的地方添加 mysql / mariadb dialect 分支。

autogenerate 局限

• autogenerate 无法检测列或表的重命名——它会输出 drop_column + add_column，导致数据丢失。重命名时请使用 op.alter_column(..., new_column_name=...) 手工编写迁移。
• autogenerate 不会输出 Compat* 类型——上述手工改写是强制的。
• autogenerate 不会推断方言专属的 upsert 逻辑；upsert 变更保留在 database/orm_crud/_bulk.py 中，不进入迁移。

TOML 存储与 localstore

database/toml_store/ 提供异步、基于 TOML 的存储助手，用于不值得建关系表的轻量运行时配置。它与 nonebot_plugin_localstore 协作：

• 文件路径通过 get_plugin_config_file()、get_plugin_data_file() 或 get_plugin_cache_file() 解析——绝不硬编码 Path("...")。
• 三个核心异步助手覆盖全部运行时场景：
  • load_toml_dict_async(path, default=..., merge_default=...)——非阻塞地读取 TOML 表。
  • write_toml_dict_file_async(path, data, schema_basename=...)——通过 tempfile + os.replace 原子覆盖 TOML 文件，可选注入 #:schema 指令。
  • ensure_toml_dict_file_async(path, default, schema_basename=...)——仅在文件缺失时用默认值创建；从不覆盖已有文件。
• 同步对应版本（load_toml_dict_sync、ensure_toml_dict_file_sync）用于 import-time 初始化。

所有写入均使用原子替换（mkstemp + aiofiles.os.replace），写入过程中崩溃不会破坏原文件。ensure_toml_dict_file_async() 仅创建缺失文件；要覆盖现有文件请使用 write_toml_dict_file_async()。运行时配置默认值必须 JSON 可序列化；写入 TOML 时以 mode="json" 转储 Pydantic 默认值。

ORM 辅助

database/orm_crud/ 拆分为三个模块：

模块	导出
_base.py	共享辅助：_combined_conditions、_get_column_map、_is_fk_constraint_violation、_orders、_validate_column_values、DatabaseError、ROWCOUNT_UNKNOWN
_single.py	create、get_one、get_or_create、update、update_or_create、delete、exists、count
_bulk.py	bulk_create、upsert、list_items、async_iterate_safe

bulk_create(..., partial=True) 使用逐行 savepoint，使失败行被跳过并上报，而非中断整批。async_iterate_safe() 以 yield_per 流式遍历大型结果集并支持异步回调，可选收集条目。

Session 归属与 handler 注入

database/orm_crud/*.py 和 repositories/*.py 中每个函数的第一位置参数都是 session: AsyncSession | async_scoped_session。这些辅助不会自己 get_session()，不会 commit，也不会 rollback —— 事务边界由调用方控制。这样既简化了测试 fixture（传 mock session 即可），又避免了 NoneBot 请求处理中的嵌套 session。

Handler 与 async_scoped_session 模式

NoneBot matcher handler 通过 nonebot_plugin_orm 导出的 async_scoped_session 类型别名获取 scoped session。Depends 已经嵌入该别名的 Annotated 元数据，因此正确的 handler 签名是纯类型注解——不要写 = Depends(async_scoped_session)：

from nonebot import require, on_command
from nonebot.adapters import Bot, Event
from nonebot_plugin_orm import async_scoped_session

require("nonebot_plugin_orm")

from ..repositories.blocklist import upsert_block  # noqa: E402 (post-require import)


@on_command("block")
async def handle_block(
    bot: Bot,
    event: Event,
    session: async_scoped_session,
) -> None:
    await upsert_block(
        session,
        platform_id="qq",
        adapter_id="~onebot.v11",
        subject_id=str(event.get_user_id()),
    )
    await session.commit()

scoped session 由 nonebot_plugin_orm 在 matcher 运行期间打开，handler 返回时自动移除。若 handler 写入了数据，返回前一定要 await session.commit() —— repository 函数不会自己 commit。

Wrapper 函数与 @wraps

包装 handler 的装饰器（例如 handle/qq/commands/common.py 中的 _permission_wrapper）MUST 使用 functools.wraps，让 inspect.signature(wrapper) 跟随被包装函数。NoneBot 通过 wrapper 的 signature 决定要注入哪些 kwargs（bot、event、session）。在 wrapper 内部通过 session = kwargs.get("session") 提取 session，不要重新 get_session()。

后台任务与 fire-and-forget helper

后台任务自管 session 生命周期，因为它们不是 NoneBot handler 依赖。services/scheduler.py 和 services/message_store.py 保留显式模式：

async with get_session() as session:
    await repository_function(session, ...)

fire-and-forget helper 若在保留 Protocol/Callable 签名的同时包装了 session-first 的 repository 函数（例如 services/llm/agent.py 的 _default_permission_resolver、services/llm/mcp_audit.py 的 _default_audit_writer），则内部自己开 scoped session。这样把缝留在 helper 内部 —— 调用方仍按 resolver(context) 调用，helper 内部满足新的 repository API。

测试 fixture

handler 测试使用 mock_session fixture：AsyncMock（用于 async session 方法）配 MagicMock（用于同步的 add / add_all）：

@pytest.fixture
def mock_session() -> Mock:
    sess = AsyncMock()
    sess.add = MagicMock()
    sess.add_all = MagicMock()
    return sess

断言 repository 调用时注意 args[0] 现在是 session（第一位置参数）。例如 mock.call_args.args[1] 是 model 实例，args[2] 是第一个用户传入参数。

阻塞路径

文件存储、TOML 解析、deepcopy、命令解析和翻译 catalog 加载会放到 worker 线程执行。不要在异步处理器中直接调用同步存储方法。

参见

消息存储专属配置与保留策略，参见消息存储。由 TOML 存储支持的运行时配置文件，参见架构概览。