104 lines
3.1 KiB
Markdown
104 lines
3.1 KiB
Markdown
# chongming-database
|
||
|
||
异步 SQLite 数据库封装，基于 `aiosqlite` + `SQLModel`（SQLAlchemy），提供简洁的初始化与会话管理接口，内置多进程并发安全。
|
||
|
||
---
|
||
|
||
## 安装
|
||
|
||
```bash
|
||
uv pip install -e utils/database
|
||
```
|
||
|
||
---
|
||
|
||
## API
|
||
|
||
### `Database` 类
|
||
|
||
```python
|
||
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 定义表
|
||
|
||
```python
|
||
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)
|
||
)
|
||
```
|
||
|
||
### 完整生命周期
|
||
|
||
```python
|
||
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()
|
||
```
|
||
|
||
### 插件中的典型用法
|
||
|
||
```python
|
||
# src/plugins/<name>/__init__.py
|
||
from pathlib import Path
|
||
from chongming_database import Database
|
||
from .global_var import set_db
|
||
|
||
async def create_workers(session, config, count, worker_manager):
|
||
db_path = Path(config.get("data_dir", "./data")) / "<name>.db"
|
||
db = Database(db_path)
|
||
await db.initialize()
|
||
set_db(db)
|
||
...
|
||
```
|
||
|
||
---
|
||
|
||
## 依赖
|
||
|
||
- `aiosqlite` — 异步 SQLite 驱动
|
||
- `sqlmodel` — SQLAlchemy + Pydantic ORM
|
||
- `sqlalchemy` — 异步引擎支持
|