blog.pyaqa.ru/app/infrastructure/repositories/post.py

"""SQLAlchemy implementation of PostRepository."""

from uuid import UUID

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domain.entities import Post
from app.domain.repositories import PostRepository
from app.domain.value_objects import Content, Slug, Title
from app.infrastructure.database.models import PostORM


class SQLAlchemyPostRepository(PostRepository):
    """SQLAlchemy implementation of Post repository."""

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

    def _to_domain(self, orm: PostORM) -> Post:
        """Convert ORM model to domain entity."""
        return Post(
            id=UUID(orm.id),
            title=Title(orm.title),
            content=Content(orm.content),
            slug=Slug(orm.slug),
            author_id=orm.author_id,
            published=orm.published,
            tags=orm.tags or [],
            created_at=orm.created_at,
            updated_at=orm.updated_at,
        )

    def _to_orm(self, post: Post) -> PostORM:
        """Convert domain entity to ORM model."""
        return PostORM(
            id=str(post.id),
            title=post.title.value,
            content=post.content.value,
            slug=post.slug.value,
            author_id=post.author_id,
            published=post.published,
            tags=post.tags,
            created_at=post.created_at,
            updated_at=post.updated_at,
        )

    async def get_by_id(self, entity_id: UUID) -> Post | None:
        """Get post by ID."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.id == str(entity_id))
        )
        orm = result.scalar_one_or_none()
        return self._to_domain(orm) if orm else None

    async def get_all(self) -> list[Post]:
        """Get all posts."""
        result = await self._session.execute(select(PostORM))
        orms = result.scalars().all()
        return [self._to_domain(orm) for orm in orms]

    async def add(self, entity: Post) -> None:
        """Add new post."""
        orm = self._to_orm(entity)
        self._session.add(orm)
        # Commit делает TransactionManager

    async def update(self, entity: Post) -> None:
        """Update existing post."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.id == str(entity.id))
        )
        orm = result.scalar_one()

        orm.title = entity.title.value
        orm.content = entity.content.value
        orm.slug = entity.slug.value
        orm.published = entity.published
        orm.tags = entity.tags
        orm.updated_at = entity.updated_at

        # Commit делает TransactionManager

    async def delete(self, entity_id: UUID) -> None:
        """Delete post by ID."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.id == str(entity_id))
        )
        orm = result.scalar_one_or_none()
        if orm:
            await self._session.delete(orm)

    async def exists(self, entity_id: UUID) -> bool:
        """Check if post exists."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.id == str(entity_id))
        )
        return result.scalar_one_or_none() is not None

    async def get_by_slug(self, slug: str) -> Post | None:
        """Get post by slug."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.slug == slug)
        )
        orm = result.scalar_one_or_none()
        return self._to_domain(orm) if orm else None

    async def get_by_author(self, author_id: str) -> list[Post]:
        """Get posts by author."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.author_id == author_id)
        )
        orms = result.scalars().all()
        return [self._to_domain(orm) for orm in orms]

    async def get_published(self) -> list[Post]:
        """Get published posts."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.published.is_(True))
        )
        orms = result.scalars().all()
        return [self._to_domain(orm) for orm in orms]

    async def get_by_tag(self, tag: str) -> list[Post]:
        """Get posts by tag."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.tags.contains([tag]))
        )
        orms = result.scalars().all()
        return [self._to_domain(orm) for orm in orms]

    async def slug_exists(self, slug: str) -> bool:
        """Check if slug exists."""
        result = await self._session.execute(
            select(PostORM).where(PostORM.slug == slug)
        )
        return result.scalar_one_or_none() is not None

    async def search(self, query: str) -> list[Post]:
        """Search posts."""
        search_pattern = f"%{query}%"
        result = await self._session.execute(
            select(PostORM).where(
                or_(
                    PostORM.title.ilike(search_pattern),
                    PostORM.content.ilike(search_pattern),
                )
            )
        )
        orms = result.scalars().all()
        return [self._to_domain(orm) for orm in orms]