from pathlib import Path
from chongming_database import Database

# SQLite（文件路径）
db = Database(Path("data/storage/chongming/app.db"))
await db.initialize()     # 创建引擎 + 自动建表

# PostgreSQL（URL）
db = Database.from_url("postgresql+asyncpg://user:pwd@127.0.0.1:5432/mydb")
await db.initialize()

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

await db.dispose()         # 关闭连接

构造方式

方式	说明
`Database(db_path)`	SQLite 文件路径，自动创建父目录（向后兼容旧用法）
`Database(url=...)`	SQLAlchemy URL（PostgreSQL），`db_path` 与 `url` 必须且只能提供一个
`Database.from_url(url)`	同上，classmethod 写法
`Database.from_config(c)`	按配置字典创建，对齐 config 中 Database TypedDict 的契约（见下）

`from_config` 契约

与各处 config 中已有的 Database TypedDict 对齐：

# SQLite
Database.from_config({"type": "sqlite", "database_path": "./data/app.db"})

# PostgreSQL（裸 postgresql:// 会自动归一化为 postgresql+asyncpg://）
Database.from_config({"type": "postgresql", "url": "postgresql://user:pwd@127.0.0.1:5432/mydb"})

type="mysql" 暂不支持，抛 NotImplementedError。

方法说明

方法	签名	说明
`__init__`	`(db_path=None, url=None)`	二选一必填，否则抛 `ValueError`；`db_path` 自动创建父目录
`initialize`	`(generate_schemas=True, echo=False, **kwargs)`	创建异步引擎，`SQLModel.metadata.create_all` 自动建表（`generate_schemas=False` 可跳过，交由外部迁移工具），构建 `sessionmaker`

PostgreSQL：initialize() 检测到目标库不存在时（如全新 PG 实例），会以相同凭据连接 postgres 维护库自动 CREATE DATABASE（多进程并发启动安全），因此无需在部署流程中手工建库；要求连接账号具备 CREATEDB 权限（默认超级用户 postgres 满足）。 | get_session | () -> AsyncSession | 返回新的异步会话；未初始化则抛 RuntimeError | | dispose | async () -> None | 关闭异步引擎并清除内部状态 |

多进程安全

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

SQLite：DDL 自动提交，用 connect() 建表；PG：DDL 事务性，用 begin() 建表。

使用示例

配合 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()

PostgreSQL 本地开发（Docker）

docker run -d --name chongming-pg \
  -e POSTGRES_PASSWORD=chongming -e POSTGRES_DB=chongming \
  -p 5432:5432 postgres:16-alpine

# 连接
# postgresql+asyncpg://postgres:chongming@127.0.0.1:5432/chongming

插件中的典型用法

推荐通过 Plugin 基类自动建库：

SQLite：db_name 指定文件名，data_dir 取自运行时配置（默认行为，现有插件零改动）
PostgreSQL：在插件 config.toml 的 [workers.config.database] 下配置 type/url，基类 init_database() 检测到 url 即走 Database.from_config 接入 PG

# config/workers/<name>/config.toml
[workers.config.database]
type = "postgresql"
url = "postgresql+asyncpg://postgres:chongming@127.0.0.1:5432/chongming"
generate_schemas = true

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

class MyPlugin(Plugin):
    type = "<name>"
    # db_name 用于 SQLite；配置了 [workers.config.database].url 时优先走 PostgreSQL

    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()

测试

uv sync --dev

# SQLite 单元测试（无需外部服务）
pytest tests/

# PostgreSQL 集成测试（需本地 PG，可通过上面的 docker 命令启动）
CHONGMING_TEST_PG_URL=postgresql+asyncpg://postgres:chongming@127.0.0.1:5432/chongming pytest tests/

未设置 CHONGMING_TEST_PG_URL 时 PG 测试自动跳过。

依赖

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

README.md Unescape Escape

chongming-database

安装

API

Database 类