baozaotumao
df48c36422
把规则从原 ppt-gen 项目泛化为通用脚手架，覆盖 19 个文件： - ddd/architecture/design-discipline/error-codes/communication-* 等： 领域术语、错误码、消息示例、扩展点、状态流统一为「订单」示范域 - agent-dag/agent.md/agent-concurrency/agent-llm-*：DAG 节点、配置、并发、 LLM 适配从「对话→预览→生成→截图→合成」改为「intake→process→validate→finalize」 - backend-db：Session 实体/状态机/ORM/仓储/checkpoint 对齐新流水线 - 移除幻灯片专属依赖与产物：agent 去 playwright/python-pptx， Dockerfile 去 Chromium 系统库，.gitignore 去 *.pptx - 保留合法通用项：ANSI（PTY 处理）、Playwright（E2E 测试工具）、Gemini CLI（真实 provider） 验证：gemini-cli/openai-api 两变体生成退出 0，生成项目零 PPT 残留， backend/agent 骨架均可 import 运行。
236 lines
7.5 KiB
Django/Jinja
236 lines
7.5 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
|
||
input_summary: str = "" # 输入摘要（示范）
|
||
work_dir: str = ""
|
||
status: SessionStatus = SessionStatus.IDLE
|
||
result_path: str | None = None
|
||
error_message: str | None = None
|
||
|
||
def mark_validating(self, result_path: str) -> None:
|
||
if self.status != SessionStatus.PROCESSING:
|
||
raise InvalidTransitionError(self.status, SessionStatus.VALIDATING)
|
||
self.result_path = result_path
|
||
self.status = SessionStatus.VALIDATING
|
||
|
||
def mark_finalizing(self) -> None:
|
||
self.status = SessionStatus.FINALIZING
|
||
|
||
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"
|
||
INTAKE = "intake"
|
||
PROCESSING = "processing"
|
||
VALIDATING = "validating"
|
||
FINALIZING = "finalizing"
|
||
DONE = "done"
|
||
ERROR = "error"
|
||
|
||
def can_recover(self) -> bool:
|
||
return self in (self.VALIDATING, self.FINALIZING, self.DONE)
|
||
```
|
||
|
||
## SessionRecord ORM 映射（db/models.py）
|
||
|
||
ORM 类仅做持久化映射，**不含任何业务逻辑**。
|
||
|
||
```python
|
||
class SessionRecord(Base):
|
||
__tablename__ = "sessions"
|
||
|
||
id: Mapped[str] = mapped_column(String(36), primary_key=True)
|
||
input_summary: Mapped[str] = mapped_column(String(500))
|
||
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)
|
||
result_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.result_path = session.result_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, input_summary=session.input_summary,
|
||
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, input_summary=r.input_summary, work_dir=r.work_dir,
|
||
status=SessionStatus(r.status),
|
||
result_path=r.result_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 / 断点恢复
|
||
|
||
中间结果成功落盘是 checkpoint 分界线：
|
||
|
||
| 断连时状态 | 能否恢复 | 恢复行为 |
|
||
|-----------|---------|---------|
|
||
| `intake` / `processing` | ❌ | 推送 `checkpoint_lost`，提示用户重新开始 |
|
||
| `validating` | ✅ | 跳过 LLM，重跑校验 |
|
||
| `finalizing` | ✅ | 中间结果已在磁盘，重跑落地 |
|
||
| `done` | ✅ | 直接推送 `result_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` 定时任务软删→硬删过期记录。
|