Merge pull request #83 from seamusic-official/SM-67

ENHANCEMENT SM-67: add DAO & PostgresDAOImplementation

Author: dmkjfs

migrations/models.py

@@ -18,7 +18,7 @@
     producers_to_beatpacks_association,
     followers_to_producers_association,
 )
-from src.app.auth.users.models import (  # noqa: F401
+from src.app.auth.users.interfaces.da.models import (  # noqa: F401
     User,
     follower_to_tag_association,
     author_to_licenses_association,

src/app/auth/users/interfaces/da/dao.py

@@ -0,0 +1,249 @@
+from dataclasses import dataclass
+from datetime import date, datetime
+from types import TracebackType
+from typing import Literal, Self
+
+from sqlalchemy import select, func
+
+from src.app.auth.artists.models import ArtistProfile
+from src.app.auth.producers.models import ProducerProfile
+from src.app.auth.users.interfaces.da.models import User
+from src.app.music.albums.interfaces.da.models import Album
+from src.app.music.squads.models import Squad
+from src.app.social.licenses.models import License
+from src.app.social.playlists.models import Playlist
+from src.app.social.tags.models import Tag
+from src.domain.auth.users.interfaces.da.dao import DAO
+from src.domain.auth.users.interfaces.da.models import AccessLevel, PremiumLevel
+from src.infrastructure.loggers import app as logger
+from src.infrastructure.postgres import PostgresSessionMixin
+
+
+@dataclass
+class PostgresDAOImplementation(DAO, PostgresSessionMixin):
+    def __init__(self) -> None:
+        super(PostgresDAOImplementation, self).__init__(table=User)
+        self.table = User
+
+    async def get_user_by_id(self, user_id: int) -> User | None:
+        logger.info("get_user_by_id DAO request")
+        response: User | None = await self.read(obj_id=user_id)
+        return response
+
+    async def get_user_by_username(self, username: str) -> User | None:
+        logger.info("get_user_by_username DAO request")
+        response: User | None = await self.run(
+            statement=select(User).group_by(User.id).filter(User.username == username),
+            method="scalar",
+        )
+        return response
+
+    async def get_users(self) -> list[User]:  # type: ignore[override]
+        logger.info("get_users DAO request")
+        response: list[User] = await self.run(
+            statement=select(User).group_by(User.id).order_by(User.updated_at),
+            method="scalars",
+        )
+        return response
+
+    async def count_users(self) -> int:
+        logger.info("count_users DAO request")
+        response: int = await self.run(
+            statement=select(func.count(User.id)),
+            method="scalar",
+        )
+        return response
+
+    async def create_user(
+        self,
+        email: str,
+        password: str,
+        username: str,
+        created_at: date,
+        updated_at: datetime,
+        access_level: Literal["user", "admin", "superuser"],
+        premium_level: Literal["none", "bot", "full"],
+        is_verified: bool,
+        artist_id: int,
+        producer_id: int,
+        licenses_ids: list[int],
+        followed_squads_ids: list[int],
+        followed_artists_ids: list[int],
+        coauthored_playlists_ids: list[int],
+        saved_playlists_ids: list[int],
+        followed_producers_ids: list[int],
+        saved_albums_ids: list[int],
+        followed_tags: list[str],
+        telegram_id: int | None = None,
+        description: str | None = None,
+        picture_url: str | None = None,
+    ) -> int:
+        logger.info("create_user DAO request")
+        user = User(
+            username=username,
+            description=description,
+            email=email,
+            password=password,
+            picture_url=picture_url,
+            created_at=created_at,
+            updated_at=updated_at,
+            access_level=access_level,
+            premium_level=premium_level,
+            telegram_id=telegram_id,
+            is_verified=is_verified,
+            artist_id=artist_id,
+            producer_id=producer_id,
+            artist_profile=await self.run(
+                statement=select(ArtistProfile).group_by(ArtistProfile.id).filter(ArtistProfile.id == artist_id),
+                method="scalar",
+            ),
+            producer_profile=await self.run(
+                statement=select(ProducerProfile).group_by(ProducerProfile.id).filter(ProducerProfile.id == producer_id),
+                method="scalar",
+            ),
+            licenses=await self.run(
+                statement=select(License).group_by(License.id).filter(License.id.in_(licenses_ids)),
+                method="scalars",
+            ) if licenses_ids else list(),
+            followed_squads=await self.run(
+                statement=select(Squad).group_by(Squad.id).filter(Squad.id.in_(followed_squads_ids)),
+                method="scalars",
+            ) if followed_squads_ids else list(),
+            followed_artists=await self.run(
+                statement=select(ArtistProfile).group_by(ArtistProfile.id).filter(ArtistProfile.id.in_(followed_artists_ids)),
+                method="scalars",
+            ) if followed_artists_ids else list(),
+            coauthored_playlists=await self.run(
+                statement=select(Playlist).group_by(Playlist.id).filter(Playlist.id.in_(coauthored_playlists_ids)),
+                method="scalars",
+            ) if coauthored_playlists_ids else list(),
+            saved_playlists=await self.run(
+                statement=select(Playlist).group_by(Playlist.id).filter(Playlist.id.in_(saved_playlists_ids)),
+                method="scalars",
+            ) if saved_playlists_ids else list(),
+            followed_producers=await self.run(
+                statement=select(ProducerProfile).group_by(ProducerProfile.id).filter(ProducerProfile.id.in_(followed_producers_ids)),
+                method="scalars",
+            ) if followed_producers_ids else list(),
+            saved_albums=await self.run(
+                statement=select(Album).group_by(Album.id).filter(Album.id.in_(saved_albums_ids)),
+                method="scalars",
+            ) if saved_albums_ids else list(),
+            followed_tags=await self.run(
+                statement=select(Tag).group_by(Tag.id).filter(Tag.name.in_(followed_tags)),
+                method="scalars",
+            ) if followed_tags else list(),
+        )
+        await self.create(user)
+        return user.id
+
+    async def update_user(
+        self,
+        user_id: int,
+        username: str | None = None,
+        description: str | None = None,
+        email: str | None = None,
+        password: str | None = None,
+        picture_url: str | None = None,
+        created_at: date | None = None,
+        updated_at: datetime | None = None,
+        access_level: AccessLevel | None = None,
+        premium_level: PremiumLevel | None = None,
+        artist_id: int | None = None,
+        producer_id: int | None = None,
+        telegram_id: int | None = None,
+        is_verified: bool | None = None,
+        licenses_ids: list[int] | None = None,
+        followed_squads_ids: list[int] | None = None,
+        followed_artists_ids: list[int] | None = None,
+        coauthored_playlists_ids: list[int] | None = None,
+        saved_playlists_ids: list[int] | None = None,
+        followed_producers_ids: list[int] | None = None,
+        saved_albums_ids: list[int] | None = None,
+        followed_tags: list[str] | None = None,
+    ) -> int:
+        logger.info("update_user DAO request")
+        user = User(**dict(filter(
+            lambda item: bool(item[1]),
+            {
+                "id": user_id,
+                "username": username if username else None,
+                "description": description if description else None,
+                "email": email if email else None,
+                "password": password if password else None,
+                "picture_url": picture_url if picture_url else None,
+                "created_at": created_at if created_at else None,
+                "updated_at": updated_at if updated_at else None,
+                "access_level": str(access_level) if access_level else None,
+                "premium_level": str(premium_level) if premium_level else None,
+                "artist_id": artist_id if artist_id else None,
+                "producer_id": producer_id if producer_id else None,
+                "telegram_id": telegram_id if telegram_id else None,
+                "is_verified": is_verified if is_verified else None,
+                "artist_profile": await self.run(
+                    statement=select(ArtistProfile).group_by(ArtistProfile.id).filter(ArtistProfile.id == artist_id),
+                    method="scalar",
+                ) if artist_id else None,
+                "profile_profile": await self.run(
+                    statement=select(ProducerProfile).group_by(ProducerProfile.id).filter(ProducerProfile.id == producer_id),
+                    method="scalar",
+                ),
+                "licenses": await self.run(
+                    statement=select(License).group_by(License.id).filter(License.id.in_(licenses_ids)),
+                    method="scalars",
+                ) if licenses_ids else list(),
+                "followed_squads": await self.run(
+                    statement=select(Squad).group_by(Squad.id).filter(Squad.id.in_(followed_squads_ids)),
+                    method="scalars",
+                ) if followed_squads_ids else list(),
+                "followed_artists": await self.run(
+                    statement=select(ArtistProfile).group_by(ArtistProfile.id).filter(ArtistProfile.id.in_(followed_artists_ids)),
+                    method="scalars",
+                ) if followed_artists_ids else list(),
+                "coauthored_playlists": await self.run(
+                    statement=select(Playlist).group_by(Playlist.id).filter(Playlist.id.in_(coauthored_playlists_ids)),
+                    method="scalars",
+                ) if coauthored_playlists_ids else list(),
+                "saved_playlists": await self.run(
+                    statement=select(Playlist).group_by(Playlist.id).filter(Playlist.id.in_(saved_playlists_ids)),
+                    method="scalars",
+                ) if saved_playlists_ids else list(),
+                "followed_producers": await self.run(
+                    statement=select(ProducerProfile).group_by(ProducerProfile.id).filter(ProducerProfile.id.in_(followed_producers_ids)),
+                    method="scalars",
+                ) if followed_producers_ids else list(),
+                "saved_albums": await self.run(
+                    statement=select(Album).group_by(Album.id).filter(Album.id.in_(saved_albums_ids)),
+                    method="scalars",
+                ) if saved_albums_ids else list(),
+                "followed_tags": await self.run(
+                    statement=select(Tag).group_by(Tag.id).filter(Tag.name.in_(followed_tags)),
+                    method="scalars",
+                ) if followed_tags else list(),
+            }.items()
+        )))
+        await self.create(user)
+        return user.id
+
+    async def delete_user(self, user_id: int) -> None:
+        logger.info("delete_user DAO request")
+        await self.delete_(obj_id=user_id)
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[Exception],
+        exc_val: Exception,
+        exc_tb: TracebackType,
+    ) -> None:
+        await self.commit()
+        await self.close()
+
+
+def get_postgres_dao_implementation() -> PostgresDAOImplementation:
+    """
+    :return: instance of PostgresDAOImplementation
+    """
+    return PostgresDAOImplementation()

src/app/auth/users/interfaces/da/models.py

@@ -0,0 +1,114 @@
+from datetime import date, datetime
+
+from sqlalchemy import Table, ForeignKey, Integer, Column
+from sqlalchemy.orm import Mapped, relationship, mapped_column
+
+from src.app.auth.artists.models import user_to_artist_association, ArtistProfile, followers_to_artists_association
+from src.app.auth.producers.models import (
+    ProducerProfile,
+    user_to_producer_association,
+    followers_to_producers_association,
+)
+from src.app.music.albums.interfaces.da.models import saver_to_albums_association, Album
+from src.app.music.squads.models import Squad, follower_to_squads_association
+from src.app.social.licenses.models import author_to_licenses_association, License
+from src.app.social.playlists.models import Playlist, author_to_playlists_association
+from src.app.social.tags.models import Tag
+from src.domain.auth.users.interfaces.da.models import BaseUserModel
+from src.infrastructure.postgres import Base
+
+follower_to_tag_association = Table(
+    "follower_to_tag_association",
+    Base.metadata,
+    Column("user_id", ForeignKey("users.id"), primary_key=True),
+    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
+)
+
+saver_to_playlists_association = Table(
+    "saver_to_playlists_association",
+    Base.metadata,
+    Column("saver_id", Integer, ForeignKey("users.id"), primary_key=True),
+    Column("playlist_id", Integer, ForeignKey("playlists.id"), primary_key=True),
+)
+
+
+class User(BaseUserModel, Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)  # type: ignore[assignment]
+    username: Mapped[str]  # type: ignore[assignment]
+    description: Mapped[str | None]  # type: ignore[assignment]
+    email: Mapped[str]  # type: ignore[assignment]
+    password: Mapped[str]  # type: ignore[assignment]
+    picture_url: Mapped[str | None]  # type: ignore[assignment]
+
+    created_at: Mapped[date]  # type: ignore[assignment]
+    updated_at: Mapped[datetime]  # type: ignore[assignment]
+
+    access_level: Mapped[str]  # type: ignore[assignment]
+    premium_level: Mapped[str]  # type: ignore[assignment]
+    telegram_id: Mapped[int | None]  # type: ignore[assignment]
+    is_verified: Mapped[bool]  # type: ignore[assignment]
+
+    artist_id: Mapped[int] = mapped_column(ForeignKey("artist_profiles.id"))  # type: ignore[assignment]
+    producer_id: Mapped[int] = mapped_column(ForeignKey("producer_profiles.id"))  # type: ignore[assignment]
+
+    artist_profile: Mapped[ArtistProfile] = relationship(
+        argument="ArtistProfile",
+        secondary=user_to_artist_association,
+        cascade="all, delete-orphan",
+        single_parent=True,
+        lazy="joined",
+    )
+    producer_profile: Mapped[ProducerProfile] = relationship(
+        argument="ProducerProfile",
+        secondary=user_to_producer_association,
+        cascade="all, delete-orphan",
+        single_parent=True,
+        lazy="joined",
+    )
+    licenses: Mapped[list[License]] = relationship(  # type: ignore[assignment]
+        argument="License",
+        secondary=author_to_licenses_association,
+        back_populates="author",
+        lazy="selectin",
+    )
+    followed_squads: Mapped[list[Squad]] = relationship(  # type: ignore[assignment]
+        argument="Squad",
+        secondary=follower_to_squads_association,
+        back_populates="followers",
+        lazy="selectin",
+    )
+    followed_artists: Mapped[list[ArtistProfile]] = relationship(  # type: ignore[assignment]
+        argument="ArtistProfile",
+        secondary=followers_to_artists_association,
+        back_populates="followers",
+        lazy="selectin",
+    )
+    coauthored_playlists: Mapped[list[Playlist]] = relationship(  # type: ignore[assignment]
+        argument="Playlist",
+        secondary=author_to_playlists_association,
+        back_populates="authors",
+        lazy="selectin",
+    )
+    saved_playlists: Mapped[list[Playlist]] = relationship(  # type: ignore[assignment]
+        argument="Playlist",
+        secondary=saver_to_playlists_association,
+        lazy="selectin",
+    )
+    followed_producers: Mapped[list[ProducerProfile]] = relationship(  # type: ignore[assignment]
+        argument="ProducerProfile",
+        secondary=followers_to_producers_association,
+        back_populates="followers",
+        lazy="selectin",
+    )
+    saved_albums: Mapped[list[Album]] = relationship(  # type: ignore[assignment]
+        argument="Album",
+        secondary=saver_to_albums_association,
+        lazy="selectin",
+    )
+    followed_tags: Mapped[list[Tag]] = relationship(  # type: ignore[assignment]
+        argument="Tag",
+        secondary=follower_to_tag_association,
+        lazy="selectin",
+    )