上一篇我们看到 Service 层用 persistence_mgr.execute_async(...) 读写数据库、用 serialize_model(...) 给前端返回脱敏后的字典。这一篇就把这层地基刨开：LangBot 如何用同一套 ORM 模型和迁移脚本，同时支撑「个人零配置 SQLite」和「企业级 PostgreSQL」两种数据库——这又是「双形态」原则在数据层的体现。难点不在建表，而在「老库怎么平滑升级」，所以本篇一半篇幅留给迁移系统。

一、这一层解决什么问题

LangBot 面向 toC 和 toB 两类场景（见第一篇）：

• 个人用户运行 uv run main.py，希望零配置——不装数据库，开箱即用。→ SQLite。
• 企业生产部署，需要并发、备份、运维生态。→ PostgreSQL。

如果为两种库各写一套代码，维护成本翻倍且极易漂移。LangBot 的答案是：业务代码只面向 SQLAlchemy 抽象，数据库差异收敛到「一个可插拔的 DatabaseManager」和「Alembic 的 batch 模式」两个点上。除此之外，整个上层完全感知不到自己在跟哪种库说话。

flowchart TB
    SVC["业务层 (Service / Stage / Manager)"]
    PM["PersistenceManager<br>execute_async / serialize_model"]
    DB["BaseDatabaseManager (抽象)"]
    SQLITE["SQLiteDatabaseManager<br>sqlite+aiosqlite"]
    PG["PostgreSQLDatabaseManager<br>postgresql+asyncpg"]
    SVC -->|只用 SQLAlchemy 语句| PM
    PM --> DB
    DB -.config.yaml database.use.-> SQLITE
    DB -.-> PG

二、数据库管理器：抽象与运行时选择

2.1 一个抽象基类 + 注册表

数据库差异被关进 BaseDatabaseManager（src/langbot/pkg/persistence/database.py:24）。它只暴露一件事：交出一个 SQLAlchemy 的 AsyncEngine。注册方式是项目里反复出现的「装饰器 + 全局列表」套路（和第二篇启动阶段、第四篇 Stage 同源）：

# src/langbot/pkg/persistence/database.py:10
preregistered_managers: list[type[BaseDatabaseManager]] = []

def manager_class(name: str) -> None:
    """Register a database manager class"""

    def decorator(cls: type[BaseDatabaseManager]) -> type[BaseDatabaseManager]:
        cls.name = name
        preregistered_managers.append(cls)
        return cls

    return decorator

class BaseDatabaseManager(abc.ABC):
    name: str
    ap: app.Application
    engine: sqlalchemy_asyncio.AsyncEngine

    @abc.abstractmethod
    async def initialize(self) -> None: ...

    def get_engine(self) -> sqlalchemy_asyncio.AsyncEngine:
        return self.engine

2.2 两个实现，差异只有一行 URL

两个具体实现各自只干一件事——按配置拼出引擎 URL。这就是整个「双库支持」在连接层的全部代码量：

# src/langbot/pkg/persistence/databases/sqlite.py:8
@database.manager_class('sqlite')
class SQLiteDatabaseManager(database.BaseDatabaseManager):
    async def initialize(self) -> None:
        db_file_path = self.ap.instance_config.data.get('database', {}).get('sqlite', {}).get('path', 'data/langbot.db')
        engine_url = f'sqlite+aiosqlite:///{db_file_path}'
        self.engine = sqlalchemy_asyncio.create_async_engine(engine_url)

# src/langbot/pkg/persistence/databases/postgresql.py:8
@database.manager_class('postgresql')
class PostgreSQLDatabaseManager(database.BaseDatabaseManager):
    async def initialize(self) -> None:
        postgresql_config = self.ap.instance_config.data.get('database', {}).get('postgresql', {})
        # host / port / user / password / database ...
        engine_url = f'postgresql+asyncpg://{user}:{password}@{host}:{port}/{database}'
        self.engine = sqlalchemy_asyncio.create_async_engine(engine_url)

注意两者都用异步驱动（aiosqlite / asyncpg），与全局 Quart 异步栈一致。运行时选哪个，由 PersistenceManager.initialize() 读 config.yaml 的 database.use 决定：

# src/langbot/pkg/persistence/mgr.py:37
database_type = self.ap.instance_config.data.get('database', {}).get('use', 'sqlite')
for manager in database.preregistered_managers:
    if manager.name == database_type:
        self.db = manager(self.ap)
        await self.db.initialize()
        break

默认值是 'sqlite'——呼应「个人零配置」的设计目标。

三、ORM 模型与 PersistenceManager

3.1 模型：朴素的 Declarative