feat(user): concrete implementation of UserRepository using Postgres

Author: dimanu-py

src/shared/infra/persistence/sqlalchemy/session_maker.py

@@ -5,13 +5,15 @@
     Base,
 )
 
+from src.shared.infra.settings import Settings
+
 
 class SessionMaker:
     _session_maker: sessionmaker[Session]
     _engine: Engine
 
-    def __init__(self, url: str) -> None:
-        self._engine = create_engine(url)
+    def __init__(self, settings: Settings) -> None:
+        self._engine = create_engine(settings.postgres_url)
         self._session_maker = sessionmaker(bind=self._engine)
 
     def get_session(self) -> Session:

src/shared/infra/settings.py

@@ -0,0 +1,15 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=(".env", ".env.prod"), extra="ignore")
+    postgres_user: str = Field(alias="POSTGRES_USER")
+    postgres_password: str = Field(alias="POSTGRES_PASSWORD")
+    postgres_db: str = Field(alias="POSTGRES_DB")
+    postgres_host: str = Field(alias="POSTGRES_HOST")
+    postgres_port: str = Field(alias="POSTGRES_PORT")
+
+    @property
+    def postgres_url(self) -> str:
+        return f"postgresql://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"

src/social/user/domain/user.py

@@ -29,3 +29,11 @@ def signup(cls, id_: str, name: str, username: str, email: str) -> "User":
             username=UserUsername(username),
             email=UserEmail(email),
         )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id_.value,
+            "name": self.name.value,
+            "username": self.username.value,
+            "email": self.email.value,
+        }

src/social/user/infra/persistence/__init__.py

@@ -0,0 +1,22 @@
+from typing import override
+
+from src.shared.infra.persistence.sqlalchemy.session_maker import SessionMaker
+from src.social.user.domain.user import User
+from src.social.user.domain.user_id import UserId
+from src.social.user.domain.user_repository import UserRepository
+from src.social.user.infra.persistence.user_model import UserModel
+
+
+class PostgresUserRepository(UserRepository):
+    _session_maker: SessionMaker
+
+    def __init__(self, session_maker: SessionMaker) -> None:
+        self._session_maker = session_maker
+
+    @override
+    async def save(self, user: User) -> None:
+        async with self._session_maker.get_session() as session:
+            user_to_save = UserModel(**user.to_dict())
+            session.add(user_to_save)
+            await session.commit()
+

src/social/user/infra/persistence/postgres_user_repository.py

@@ -0,0 +1,12 @@
+from src.shared.infra.persistence.sqlalchemy.base import Base
+from sqlalchemy import UUID, String
+from sqlalchemy.orm import mapped_column, Mapped
+
+
+class UserModel(Base):
+    __tablename__ = "users"
+
+    id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    username: Mapped[str] = mapped_column(String, nullable=False, unique=True)
+    email: Mapped[str] = mapped_column(String, nullable=False, unique=True)

src/social/user/infra/persistence/user_model.py

@@ -0,0 +1,30 @@
+import pytest
+
+from src.shared.infra.persistence.sqlalchemy.session_maker import SessionMaker
+from src.shared.infra.settings import Settings
+from src.social.user.infra.persistence.postgres_user_repository import (
+    PostgresUserRepository,
+)
+from src.social.user.infra.persistence.user_model import UserModel
+from tests.social.user.domain.user_mother import UserMother
+
+
+@pytest.mark.integration
+class TestPostgresUserRepository:
+    @classmethod
+    def setup_class(cls) -> None:
+        cls.session_maker = SessionMaker(settings=Settings())
+        cls.session_maker.create_tables()
+
+    @classmethod
+    def teardown_class(cls) -> None:
+        with cls.session_maker.get_session() as session:
+            session.query(UserModel).delete()
+            session.commit()
+
+    @pytest.mark.asyncio
+    async def test_should_save_user(self) -> None:
+        repository = PostgresUserRepository(session_maker=self.session_maker)
+        user = UserMother.any()
+
+        await repository.save(user)