xiaoxia-saas/packages/adapters/sqlalchemy_impl/template_version_repository.py

"""SQLAlchemy implementation of EditTemplateVersionRepository."""

from __future__ import annotations


from sqlalchemy.orm import Session

from packages.domain.template_version import EditTemplateVersion


class SQLAlchemyTemplateVersionRepository:
    """模板版本仓储实现（SQLAlchemy）。"""

    def __init__(self, db: Session) -> None:
        self._db = db

    def create(self, version: EditTemplateVersion) -> EditTemplateVersion:
        """保存新版本快照"""
        from packages.adapters.sqlalchemy_impl.models import EditTemplateVersionModel

        model = EditTemplateVersionModel(
            id=version.id,
            template_id=version.template_id,
            version=version.version,
            name=version.name,
            editing_mode=version.editing_mode,
            config=version.config,
            clip_configs=version.clip_configs,
            change_note=version.change_note,
            published_by=version.published_by,
            created_at=version.created_at,
        )
        self._db.add(model)
        self._db.flush()
        return version

    def get_by_version(self, template_id: str, version: int) -> EditTemplateVersion | None:
        """按版本号获取快照"""
        from packages.adapters.sqlalchemy_impl.models import EditTemplateVersionModel

        model = (
            self._db.query(EditTemplateVersionModel)
            .filter(
                EditTemplateVersionModel.template_id == template_id,
                EditTemplateVersionModel.version == version,
            )
            .first()
        )
        if model is None:
            return None
        return self._to_entity(model)

    def list_by_template(self, template_id: str, limit: int = 50) -> list[EditTemplateVersion]:
        """列出模板的所有历史版本（按版本号倒序）"""
        from packages.adapters.sqlalchemy_impl.models import EditTemplateVersionModel

        models = (
            self._db.query(EditTemplateVersionModel)
            .filter(EditTemplateVersionModel.template_id == template_id)
            .order_by(EditTemplateVersionModel.version.desc())
            .limit(limit)
            .all()
        )
        return [self._to_entity(m) for m in models]

    def _to_entity(self, model) -> EditTemplateVersion:
        return EditTemplateVersion(
            id=model.id,
            template_id=model.template_id,
            version=model.version,
            name=model.name or "",
            editing_mode=model.editing_mode or "one_take",
            config=model.config or {},
            clip_configs=model.clip_configs or [],
            change_note=model.change_note or "",
            published_by=model.published_by or "",
            created_at=model.created_at,
        )