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

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import ProjectModel
from packages.domain import Project


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

    def _to_entity(self, model: ProjectModel) -> Project:
        return Project(
            id=model.id,
            owner_user_id=model.owner_user_id,
            name=model.name,
            description=model.description,
            shared_users=model.shared_users or [],
            created_at=model.created_at,
        )

    def save(self, project: Project) -> Project:
        """保存项目（创建或更新）"""
        existing = self.session.query(ProjectModel).filter(ProjectModel.id == project.id).first()
        if existing:
            existing.owner_user_id = project.owner_user_id
            existing.name = project.name
            existing.description = project.description
            existing.shared_users = project.shared_users
        else:
            model = ProjectModel(
                id=project.id,
                owner_user_id=project.owner_user_id,
                name=project.name,
                description=project.description,
                shared_users=project.shared_users,
                created_at=project.created_at,
            )
            self.session.add(model)
        self.session.commit()
        return project

    def find_by_id(self, project_id: str) -> Project | None:
        model = self.session.query(ProjectModel).filter(ProjectModel.id == project_id).first()
        if model is None:
            return None
        return self._to_entity(model)

    def find_by_owner_user_id(self, owner_user_id: str) -> list[Project]:
        """根据所有者用户 ID 查找项目"""
        models = self.session.query(ProjectModel).filter(ProjectModel.owner_user_id == owner_user_id).all()
        return [self._to_entity(model) for model in models]

    def find_accessible_projects(self, user_id: str) -> list[Project]:
        """查找用户可访问的所有项目（自己拥有的 + 被共享的）"""
        from sqlalchemy import cast, or_
        from sqlalchemy.dialects.postgresql import JSONB

        models = (
            self.session.query(ProjectModel)
            .filter(
                or_(ProjectModel.owner_user_id == user_id, cast(ProjectModel.shared_users, JSONB).contains([user_id]))
            )
            .all()
        )
        return [self._to_entity(model) for model in models]

    def count_by_owner(self, owner_user_id: str) -> int:
        """统计用户的项目数量"""
        return self.session.query(ProjectModel).filter(ProjectModel.owner_user_id == owner_user_id).count()

    def delete(self, project_id: str) -> bool:
        """删除项目"""
        model = self.session.query(ProjectModel).filter(ProjectModel.id == project_id).first()
        if model is None:
            return False
        self.session.delete(model)
        self.session.commit()
        return True