baozaotumao
39fe248f9f
- 新增 .claude/skills/new-project/SKILL.md（标准 skill 格式） - install-skill.sh 安装时组装 SKILL.md + verify.sh，单一源无重复 - verify.sh 改用索引数组替代 declare -A，兼容 macOS bash 3.2 - log_pass/log_skip 显式 return 0，避免 set -e 下非 verbose 误退出 - README 改为 skill 优先；补全 copier.yml/scripts/template 入库
246 lines
8.0 KiB
Django/Jinja
246 lines
8.0 KiB
Django/Jinja
---
|
||
paths: ["backend/**"]
|
||
---
|
||
|
||
# backend-db
|
||
|
||
数据库（{% if database == 'sqlite' %}SQLite{% else %}PostgreSQL{% endif %} + SQLAlchemy async）、Session 模型、Alembic 迁移、Checkpoint 断点恢复与会话保留。
|
||
|
||
## 数据库配置
|
||
|
||
{% if database == 'sqlite' %}
|
||
### 文件位置
|
||
|
||
```
|
||
~/{{ project_slug }}/{{ project_slug }}.db # 生产
|
||
:memory: # 测试（conftest.py 中覆盖）
|
||
```
|
||
|
||
### 引擎配置（db/engine.py）
|
||
|
||
```python
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||
|
||
engine = create_async_engine(
|
||
f"sqlite+aiosqlite:///{settings.db_path}",
|
||
echo=settings.env == "dev",
|
||
connect_args={"check_same_thread": False},
|
||
)
|
||
|
||
# WAL 模式：提升并发读性能
|
||
@event.listens_for(engine.sync_engine, "connect")
|
||
def set_wal_mode(dbapi_conn, _):
|
||
dbapi_conn.execute("PRAGMA journal_mode=WAL")
|
||
|
||
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||
```
|
||
|
||
SQLite 无需连接池配置——`aiosqlite` 内部以串行化方式处理并发写。
|
||
{% else %}
|
||
### 连接字符串
|
||
|
||
```
|
||
# .env
|
||
DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/{{ project_slug }}
|
||
```
|
||
|
||
### 引擎配置（db/engine.py）
|
||
|
||
```python
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||
|
||
engine = create_async_engine(
|
||
settings.database_url,
|
||
echo=settings.env == "dev",
|
||
pool_size=10,
|
||
max_overflow=20,
|
||
pool_pre_ping=True, # 连接健康检查，防止使用断开的连接
|
||
pool_recycle=3600, # 1 小时回收连接，防止 idle timeout
|
||
)
|
||
|
||
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||
```
|
||
|
||
### PostgreSQL 迁移注意事项
|
||
|
||
- 大表加列（NOT NULL）需提供服务端默认值，避免全表 lock
|
||
- 建索引用 `CREATE INDEX CONCURRENTLY`（非阻塞）
|
||
- 枚举类型变更需显式 `ALTER TYPE`，Alembic autogenerate 不自动处理
|
||
{% endif %}
|
||
|
||
## Session 聚合根（domain/entities.py）
|
||
|
||
`Session` 是领域核心实体，**纯 Python，零框架/IO 依赖**。业务规则（状态迁移、恢复判断）封装在实体方法内，不外泄到 service 层。
|
||
|
||
```python
|
||
from dataclasses import dataclass
|
||
from .value_objects import SessionStatus
|
||
|
||
@dataclass
|
||
class Session:
|
||
id: str
|
||
topic: str
|
||
language: str = "中文"
|
||
style: str | None = None
|
||
image_source: str | None = None
|
||
work_dir: str = ""
|
||
status: SessionStatus = SessionStatus.IDLE
|
||
html_path: str | None = None
|
||
pptx_path: str | None = None
|
||
error_message: str | None = None
|
||
|
||
def mark_screenshotting(self, html_path: str) -> None:
|
||
if self.status != SessionStatus.GENERATING:
|
||
raise InvalidTransitionError(self.status, SessionStatus.SCREENSHOTTING)
|
||
self.html_path = html_path
|
||
self.status = SessionStatus.SCREENSHOTTING
|
||
|
||
def mark_synthesizing(self, pptx_path: str) -> None:
|
||
self.pptx_path = pptx_path
|
||
self.status = SessionStatus.SYNTHESIZING
|
||
|
||
def mark_done(self) -> None:
|
||
self.status = SessionStatus.DONE
|
||
|
||
def mark_error(self, message: str) -> None:
|
||
self.error_message = message
|
||
self.status = SessionStatus.ERROR
|
||
|
||
def can_recover(self) -> bool:
|
||
return self.status.can_recover()
|
||
```
|
||
|
||
`SessionStatus` 值对象（`domain/value_objects.py`）：
|
||
|
||
```python
|
||
from enum import StrEnum
|
||
|
||
class SessionStatus(StrEnum):
|
||
IDLE = "idle"
|
||
DIALOG = "dialog"
|
||
GENERATING = "generating"
|
||
SCREENSHOTTING = "screenshotting"
|
||
SYNTHESIZING = "synthesizing"
|
||
DONE = "done"
|
||
ERROR = "error"
|
||
|
||
def can_recover(self) -> bool:
|
||
return self in (self.SCREENSHOTTING, self.SYNTHESIZING, self.DONE)
|
||
```
|
||
|
||
## SessionRecord ORM 映射（db/models.py）
|
||
|
||
ORM 类仅做持久化映射，**不含任何业务逻辑**。
|
||
|
||
```python
|
||
class SessionRecord(Base):
|
||
__tablename__ = "sessions"
|
||
|
||
id: Mapped[str] = mapped_column(String(36), primary_key=True)
|
||
topic: Mapped[str] = mapped_column(String(500))
|
||
style: Mapped[str | None] = mapped_column(String(100))
|
||
language: Mapped[str] = mapped_column(String(20), default="中文")
|
||
image_source: Mapped[str | None] = mapped_column(String(50))
|
||
work_dir: Mapped[str] = mapped_column(String(500))
|
||
status: Mapped[str] = mapped_column(String(30), default="idle")
|
||
error_message: Mapped[str | None] = mapped_column(Text)
|
||
html_path: Mapped[str | None] = mapped_column(String(500))
|
||
pptx_path: Mapped[str | None] = mapped_column(String(500))
|
||
checkpoint_at: Mapped[datetime | None] = mapped_column(DateTime)
|
||
created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())
|
||
updated_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
|
||
completed_at: Mapped[datetime | None] = mapped_column(DateTime)
|
||
```
|
||
|
||
## SessionRepository 适配器（db/repositories.py）
|
||
|
||
`SQLAlchemySessionRepository` 实现 `domain/ports.py` 的 `SessionRepository` 端口，负责 ORM ↔ 领域实体的互转。调用方（services）对 SQLAlchemy 零感知。
|
||
|
||
```python
|
||
from typing import override
|
||
from domain.ports import SessionRepository
|
||
from domain.entities import Session
|
||
from .models import SessionRecord
|
||
|
||
class SQLAlchemySessionRepository(SessionRepository):
|
||
def __init__(self, db: AsyncSession) -> None:
|
||
self._db = db
|
||
|
||
@override
|
||
async def get(self, session_id: str) -> Session | None:
|
||
record = await self._db.get(SessionRecord, session_id)
|
||
return self._to_entity(record) if record else None
|
||
|
||
@override
|
||
async def save(self, session: Session) -> None:
|
||
record = await self._db.get(SessionRecord, session.id)
|
||
record.status = session.status
|
||
record.html_path = session.html_path
|
||
record.pptx_path = session.pptx_path
|
||
record.error_message = session.error_message
|
||
record.checkpoint_at = func.now()
|
||
await self._db.flush()
|
||
|
||
@override
|
||
async def create(self, session: Session) -> None:
|
||
record = SessionRecord(
|
||
id=session.id, topic=session.topic, language=session.language,
|
||
style=session.style, image_source=session.image_source,
|
||
work_dir=session.work_dir, status=session.status,
|
||
)
|
||
self._db.add(record)
|
||
await self._db.flush()
|
||
|
||
def _to_entity(self, r: SessionRecord) -> Session:
|
||
return Session(
|
||
id=r.id, topic=r.topic, language=r.language,
|
||
style=r.style, image_source=r.image_source, work_dir=r.work_dir,
|
||
status=SessionStatus(r.status),
|
||
html_path=r.html_path, pptx_path=r.pptx_path,
|
||
error_message=r.error_message,
|
||
)
|
||
```
|
||
|
||
## 迁移工作流（Alembic）
|
||
|
||
```bash
|
||
uv run alembic revision --autogenerate -m "描述变更"
|
||
uv run alembic upgrade head
|
||
uv run alembic downgrade -1
|
||
```
|
||
|
||
**规则：禁止直接修改已提交的迁移脚本。每次 model 变更必须生成新迁移脚本。**
|
||
|
||
## Checkpoint / 断点恢复
|
||
|
||
`slides.html` 成功落盘是 checkpoint 分界线：
|
||
|
||
| 断连时状态 | 能否恢复 | 恢复行为 |
|
||
|-----------|---------|---------|
|
||
| `dialog` / `generating` | ❌ | 推送 `checkpoint_lost`，提示用户重新开始 |
|
||
| `screenshotting` | ✅ | 跳过 LLM，重跑截图 |
|
||
| `synthesizing` | ✅ | 截图已在磁盘，重跑合成 |
|
||
| `done` | ✅ | 直接推送 `html_ready` + `pptx_ready` |
|
||
|
||
## DB 依赖注入（core/deps.py）
|
||
|
||
路由层通过 `get_session_repo` 获得 `SessionRepository` 实例，**不直接依赖 `get_db`**：
|
||
|
||
```python
|
||
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
||
async with AsyncSessionLocal() as session:
|
||
try:
|
||
yield session
|
||
await session.commit()
|
||
except Exception:
|
||
await session.rollback()
|
||
raise
|
||
|
||
async def get_session_repo(db: AsyncSession = Depends(get_db)) -> SessionRepository:
|
||
return SQLAlchemySessionRepository(db)
|
||
```
|
||
|
||
## 会话保留（DB_RETENTION_DAYS）
|
||
|
||
会话记录保留 `DB_RETENTION_DAYS`（默认 30 天），`lifespan` 定时任务软删→硬删过期记录。
|