--- 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` 定时任务软删→硬删过期记录。