Merge pull request #84 from seamusic-official/SM-68

ENHANCEMENT SM-68: add MAO & S3MAOImplementation

Author: dmkjfs

src/app/auth/users/interfaces/ma/mao.py

@@ -0,0 +1,37 @@
+from dataclasses import dataclass
+from types import TracebackType
+from typing import Self
+
+from src.domain.auth.users.interfaces.ma.mao import MAO
+from src.infrastructure.loggers import app as logger
+from src.infrastructure.s3 import S3SessionMixin, unique_filename, get_file_stream
+
+
+@dataclass
+class S3MAOImplementation(MAO, S3SessionMixin):
+    async def update_profile_picture(self, data: bytes, user_id: int) -> str:
+        logger.info("update_profile_picture MAO request")
+        cover_url = await self.update(
+            path="/users/",
+            filename=unique_filename(str(user_id)),
+            file_stream=get_file_stream(data=data),
+        )
+        return cover_url
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[Exception] | None = None,
+        exc_val: Exception | None = None,
+        exc_tb: TracebackType | None = None,
+    ) -> None:
+        pass
+
+
+def get_s3_mao_implementation() -> S3MAOImplementation:
+    """
+    :return S3MAOImplementation: an instance of S3 interface session
+    """
+    return S3MAOImplementation()

src/app/music/albums/interfaces/ma/mao.py

@@ -1,12 +1,14 @@
+from dataclasses import dataclass
 from types import TracebackType
 from typing import Self
 
 from src.domain.music.albums.interfaces.ma.mao import MAO
 from src.infrastructure.loggers import app as logger
-from src.infrastructure.s3 import Session, unique_filename, get_file_stream
+from src.infrastructure.s3 import S3SessionMixin, unique_filename, get_file_stream
 
 
-class S3MAOImplementation(MAO, Session):
+@dataclass
+class S3MAOImplementation(MAO, S3SessionMixin):
     async def update_cover(self, data: bytes, album_id: int) -> str:
         logger.info("update_cover MAO request")
         cover_url = await self.update(
@@ -29,4 +31,7 @@ async def __aexit__(
 
 
 def get_s3_mao_implementation() -> S3MAOImplementation:
+    """
+    :return S3MAOImplementation: an instance of S3 interface session
+    """
     return S3MAOImplementation()

src/domain/auth/users/interfaces/ma/mao.py

@@ -0,0 +1,49 @@
+from abc import abstractmethod
+from dataclasses import dataclass
+from types import TracebackType
+from typing import Self
+
+from src.domain.auth.users.interfaces.base import BaseInterface
+
+
+@dataclass
+class MAO(BaseInterface):
+    """
+    Media Access Object (MAO) is an abstract class created
+    to define and describe necessary functions for files acess.
+
+    It's used in services' layer to manage files.
+
+    As an abstraction, MAO cannot be used directly - that will
+    cause `NotImplementedError` and crash the application. So,
+    to provide the files access, a MAO implementation is required.
+
+    MAO implementation is a MAO's subclass designed in a unique way
+    for an exact type of media storage. It can only be used for
+    files operations with data in that type of storage. The implementation
+    must be provided to the service's layer via factory to avoid using
+    globals.
+    """
+
+    @abstractmethod
+    async def update_profile_picture(self, data: bytes, user_id: int) -> str:
+        """
+        Changes user's profile picture
+
+        :param user_id: user's numeric identificator
+        :param data: file data in bytes format
+        :return: cover's URL
+        :raise NotImplementedError: when called directly by abstract class instance
+        """
+        raise NotImplementedError
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[Exception] | None = None,
+        exc_val: Exception | None = None,
+        exc_tb: TracebackType | None = None,
+    ) -> None:
+        pass