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

"""封面模板 SQLAlchemy 仓储实现。"""

from sqlalchemy import or_
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import CoverTemplateModel
from packages.domain.cover_template import CoverTemplate


class SQLAlchemyCoverTemplateRepository:
    """封面模板仓储实现。"""

    def __init__(self, session: Session):
        self.session = session

    def create(self, template: CoverTemplate) -> CoverTemplate:
        model = CoverTemplateModel(
            id=template.id,
            user_id=template.user_id,
            name=template.name,
            thumbnail_url=template.thumbnail_url,
            is_system=template.is_system,
            config=template.config,
            created_at=template.created_at,
            updated_at=template.updated_at,
        )
        self.session.add(model)
        self.session.commit()
        return template

    def get(self, template_id: str) -> CoverTemplate | None:
        model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
        if model is None:
            return None
        return self._to_domain(model)

    def list_for_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[CoverTemplate]:
        """列出用户可见的模板：系统模板 + 用户自己的模板。"""
        models = (
            self.session.query(CoverTemplateModel)
            .filter(
                or_(
                    CoverTemplateModel.is_system == True,  # noqa: E712
                    CoverTemplateModel.user_id == user_id,
                )
            )
            .order_by(CoverTemplateModel.is_system.desc(), CoverTemplateModel.created_at.desc())
            .offset(skip)
            .limit(limit)
            .all()
        )
        return [self._to_domain(m) for m in models]

    def count_for_user(self, user_id: str) -> int:
        return (
            self.session.query(CoverTemplateModel)
            .filter(
                or_(
                    CoverTemplateModel.is_system == True,  # noqa: E712
                    CoverTemplateModel.user_id == user_id,
                )
            )
            .count()
        )

    def update(self, template: CoverTemplate) -> CoverTemplate:
        model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template.id).first()
        if model is None:
            raise ValueError(f"模板 {template.id} 不存在")
        model.name = template.name
        model.thumbnail_url = template.thumbnail_url
        model.config = template.config
        model.updated_at = template.updated_at
        self.session.commit()
        return template

    def delete(self, template_id: str) -> bool:
        model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
        if model is None:
            return False
        self.session.delete(model)
        self.session.commit()
        return True

    def list_system_templates(self) -> list[CoverTemplate]:
        models = (
            self.session.query(CoverTemplateModel)
            .filter(CoverTemplateModel.is_system == True)  # noqa: E712
            .order_by(CoverTemplateModel.created_at)
            .all()
        )
        return [self._to_domain(m) for m in models]

    @staticmethod
    def _to_domain(model: CoverTemplateModel) -> CoverTemplate:
        return CoverTemplate(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            thumbnail_url=model.thumbnail_url,
            is_system=model.is_system,
            config=model.config or {},
            created_at=model.created_at,
            updated_at=model.updated_at,
        )