chongming-database

异步 SQLite 数据库封装，基于 aiosqlite + SQLModel（SQLAlchemy），提供简洁的初始化与会话管理接口，内置多进程并发安全。

安装

uv pip install -e utils/database

API

`Database` 类

from pathlib import Path
from chongming_database import Database

db = Database(Path("data/storage/chongming/app.db"))
await db.initialize()     # 创建引擎 + 自动建表

async with db.get_session() as session:
    result = await session.execute(...)

await db.dispose()         # 关闭连接

方法说明

方法	签名	说明
`__init__`	`(db_path: Path)`	传入 SQLite 文件路径，自动创建父目录
`initialize`	`async () -> None`	创建异步引擎（`sqlite+aiosqlite:///`），`SQLModel.metadata.create_all` 自动建表，构建 `sessionmaker`（`expire_on_commit=False`）
`get_session`	`() -> AsyncSession`	返回新的异步会话；未初始化则抛 `RuntimeError`
`dispose`	`async () -> None`	关闭异步引擎并清除内部状态

多进程安全

initialize() 使用 checkfirst=True 检查表是否存在，并捕获 OperationalError("already exists") 作为兜底，确保多个进程并发启动时不会因建表冲突而崩溃。

使用示例

配合 SQLModel 定义表

from sqlmodel import SQLModel, Field
from datetime import datetime, timezone

class User(SQLModel, table=True):
    id: str = Field(primary_key=True)
    username: str = Field(max_length=50, unique=True, index=True)
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

完整生命周期

from chongming_database import Database
from pathlib import Path

db = Database(Path("./data/myapp.db"))
await db.initialize()

try:
    async with db.get_session() as session:
        user = User(id="u1", username="john")
        session.add(user)
        await session.commit()
finally:
    await db.dispose()

插件中的典型用法

推荐通过 Plugin 基类自动建库（db_name 指定文件名，data_dir 取自运行时配置）， on_init 中即可拿到 self.db：

# src/plugins/<name>/__init__.py
from chongming_worker.plugin import Plugin

class MyPlugin(Plugin):
    type = "<name>"
    db_name = "<name>.db"   # 基类自动在 data_dir 下建库并建表

    async def on_init(self):
        db = self.db   # 已由基类 init_database() 初始化
        self.set_var("db", db)   # 发布到 global_var 供业务代码读取
        ...

# cleanup 时基类自动 dispose（无需手写）

不依赖基类时也可手动使用：

from pathlib import Path
from chongming_database import Database

db_path = Path(config.get("data_dir", "./data")) / "<name>.db"
db = Database(db_path)
await db.initialize()

依赖

aiosqlite — 异步 SQLite 驱动
sqlmodel — SQLAlchemy + Pydantic ORM
sqlalchemy — 异步引擎支持

3.6 KiB Raw Blame History Unescape Escape