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:
# 使用 SQLModel / SQLAlchemy 进行查询
result = await session.execute(...)
...
# 关闭连接
await db.dispose()
方法说明
| 方法 | 签名 | 说明 |
|---|---|---|
__init__ |
(db_path: Path) |
传入 SQLite 文件路径，自动创建父目录 |
initialize |
async () -> None |
创建异步 SQLAlchemy 引擎（sqlite+aiosqlite:///），执行 SQLModel.metadata.create_all 自动建表，构建 sessionmaker（expire_on_commit=False） |
get_session |
() -> AsyncSession |
返回一个新的异步会话。如果 initialize() 未调用则抛出 RuntimeError |
dispose |
async () -> None |
关闭异步引擎并清除内部状态 |
使用示例
配合 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()
依赖
aiosqlite— 异步 SQLite 驱动sqlmodel— SQLAlchemy + Pydantic 融合 ORM