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

"""SQLAlchemy implementation of RecipeRepository."""

from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import RecipeItemModel, RecipeModel
from packages.domain.recipe import Recipe, RecipeItem


class SQLAlchemyRecipeRepository:
    """SQLAlchemy 配方仓储"""

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

    def list_by_user(
        self,
        user_id: str,
        *,
        skip: int = 0,
        limit: int = 50,
    ) -> list[Recipe]:
        models = (
            self.session.query(RecipeModel)
            .filter(
                RecipeModel.user_id == user_id,
                RecipeModel.is_active.is_(True),
            )
            .order_by(RecipeModel.created_at.desc())
            .offset(skip)
            .limit(limit)
            .all()
        )
        recipes = [self._model_to_entity(m) for m in models]
        # Load items for each recipe
        for recipe in recipes:
            recipe.items = self.list_items(recipe.id)
        return recipes

    def get(self, recipe_id: str, user_id: str) -> Optional[Recipe]:
        model = (
            self.session.query(RecipeModel)
            .filter(
                RecipeModel.id == recipe_id,
                RecipeModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        recipe = self._model_to_entity(model)
        recipe.items = self.list_items(recipe.id)
        return recipe

    def create(self, recipe: Recipe) -> Recipe:
        model = RecipeModel(
            id=recipe.id,
            user_id=recipe.user_id,
            name=recipe.name,
            description=recipe.description,
            template_id=recipe.template_id,
            generation_params=recipe.generation_params,
            is_active=recipe.is_active,
            extra_meta=recipe.metadata_,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        result = self._model_to_entity(model)
        result.items = recipe.items
        return result

    def update(self, recipe: Recipe) -> Recipe:
        model = (
            self.session.query(RecipeModel)
            .filter(
                RecipeModel.id == recipe.id,
                RecipeModel.user_id == recipe.user_id,
            )
            .first()
        )
        if model is None:
            raise ValueError(f"Recipe {recipe.id} not found")
        model.name = recipe.name
        model.description = recipe.description
        model.template_id = recipe.template_id
        model.generation_params = recipe.generation_params
        model.is_active = recipe.is_active
        model.extra_meta = recipe.metadata_
        self.session.commit()
        self.session.refresh(model)
        result = self._model_to_entity(model)
        result.items = recipe.items
        return result

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

    def count_by_user(self, user_id: str, is_active: bool = True) -> int:
        return (
            self.session.query(RecipeModel)
            .filter(
                RecipeModel.user_id == user_id,
                RecipeModel.is_active == is_active,
            )
            .count()
        )

    def list_items(self, recipe_id: str) -> list[RecipeItem]:
        models = (
            self.session.query(RecipeItemModel)
            .filter(RecipeItemModel.recipe_id == recipe_id)
            .order_by(RecipeItemModel.position)
            .all()
        )
        return [self._item_model_to_entity(m) for m in models]

    def create_items(self, items: list[RecipeItem]) -> list[RecipeItem]:
        for item in items:
            model = RecipeItemModel(
                id=item.id,
                recipe_id=item.recipe_id,
                item_type=item.item_type,
                item_id=item.item_id,
                position=item.position,
                extra_meta=item.metadata_,
            )
            self.session.add(model)
        self.session.commit()
        return items

    def delete_items_by_recipe(self, recipe_id: str) -> int:
        count = self.session.query(RecipeItemModel).filter(RecipeItemModel.recipe_id == recipe_id).delete()
        self.session.commit()
        return count

    @staticmethod
    def _model_to_entity(model: RecipeModel) -> Recipe:
        return Recipe(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            description=model.description or "",
            template_id=model.template_id or "",
            generation_params=model.generation_params or {},
            is_active=model.is_active,
            metadata_=model.extra_meta or {},
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    @staticmethod
    def _item_model_to_entity(model: RecipeItemModel) -> RecipeItem:
        return RecipeItem(
            id=model.id,
            recipe_id=model.recipe_id,
            item_type=model.item_type,
            item_id=model.item_id,
            position=model.position or 0,
            metadata_=model.extra_meta or {},
        )