feat(sql): implementing a SqlTaskRepository

app/application.py

@@ -7,6 +7,7 @@
     ORGANIZATION,
 )
 from app.fastapi import fastapi_configurer
+from app.petisco.configurers import configurers
 from app.petisco.dependencies import dependencies_provider
 
 application = FastApiApplication(
@@ -15,5 +16,6 @@
     organization=ORGANIZATION,
     deployed_at=APPLICATION_LATEST_DEPLOY,
     dependencies_provider=dependencies_provider,
+    configurers=configurers,
     fastapi_configurer=fastapi_configurer,
 )

app/petisco/configurers.py

@@ -0,0 +1,26 @@
+import os
+
+from petisco import ApplicationConfigurer, Databases
+from petisco.extra.sqlalchemy import MySqlConnection, SqlDatabase, SqliteConnection
+
+DATABASE_NAME = "sql-tasks"
+ROOT_PATH = os.path.abspath(os.path.dirname(__file__))
+SQL_SERVER = os.getenv("SQL_SERVER", "sqlite")
+
+
+class DatabasesConfigurer(ApplicationConfigurer):
+    def execute(self, testing: bool = True) -> None:
+        if testing or (SQL_SERVER == "sqlite"):
+            test_db_filename = "tasks.db"
+            connection = SqliteConnection.create("sqlite", test_db_filename)
+        else:
+            connection = MySqlConnection.from_environ()
+
+        sql_database = SqlDatabase(name=DATABASE_NAME, connection=connection)
+
+        databases = Databases()
+        databases.add(sql_database)
+        databases.initialize()
+
+
+configurers = [DatabasesConfigurer()]

app/petisco/dependencies.py

@@ -6,9 +6,10 @@
 from app.src.task.label.infrastructure.fake_task_labeler import FakeTaskLabeler
 from app.src.task.label.infrastructure.size_task_labeler import SizeTaskLabeler
 from app.src.task.shared.domain.task import Task
-from app.src.task.shared.infrastructure.folder_crud_repository import (
-    FolderTaskCrudRepository,
+from app.src.task.shared.infrastructure.folder_task_repository import (
+    FolderTaskRepository,
 )
+from app.src.task.shared.infrastructure.sql.sql_task_repository import SqlTaskRepository
 
 
 def dependencies_provider() -> list[Dependency]:
@@ -17,12 +18,10 @@ def dependencies_provider() -> list[Dependency]:
             CrudRepository,
             alias="task_repository",
             envar_modifier="TASK_REPOSITORY_TYPE",
-            strict=False,  # This should be strict due to this Bug when inherit from Generic (InmemoryCrudRepository[Task]) https://github.com/alice-biometrics/petisco/issues/356
             builders={
                 "default": Builder(InmemoryCrudRepository[Task]),
-                "folder": Builder(
-                    FolderTaskCrudRepository, folder="folder_task_database"
-                ),
+                "sql": Builder(SqlTaskRepository),
+                "folder": Builder(FolderTaskRepository, folder="folder_task_database"),
             },
         )
     ]
@@ -36,6 +35,7 @@ def dependencies_provider() -> list[Dependency]:
             },
         ),
     ]
+
     message_dependencies = get_rabbitmq_message_dependencies(
         ORGANIZATION, APPLICATION_NAME
     )

app/src/task/shared/infrastructure/folder_crud_repository.py → app/src/task/shared/infrastructure/folder_task_repository.py

@@ -20,7 +20,7 @@
 from app.src.task.shared.domain.task import Task
 
 
-class FolderTaskCrudRepository(CrudRepository[Task]):
+class FolderTaskRepository(CrudRepository[Task]):
     def __init__(self, folder: str):
         self.folder = folder
         os.makedirs(self.folder, exist_ok=True)

app/src/task/shared/infrastructure/sql/sql_task.py

@@ -0,0 +1,29 @@
+from petisco import SqlBase
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.orm import Mapped
+
+from app.src.task.shared.domain.task import Task
+
+
+class SqlTask(SqlBase[Task]):
+
+    __tablename__ = "Task"
+
+    id: Mapped[int] = Column(Integer, primary_key=True)
+
+    aggregate_id: Mapped[str] = Column(String(36))
+    name: Mapped[str] = Column(String(50))
+    description: Mapped[str] = Column(String(200))
+    # created_at: Mapped[datetime] = Column(String(200)))
+    # labels: list[str] | None = list()
+
+    def to_domain(self) -> Task:
+        return Task(
+            name=self.name, description=self.description, aggregate_id=self.aggregate_id
+        )
+
+    @staticmethod
+    def from_domain(task: Task) -> "SqlTask":
+        return SqlTask(
+            name=task.name, description=task.description, aggregate_id=task.aggregate_id
+        )

app/src/task/shared/infrastructure/sql/sql_task_repository.py

@@ -0,0 +1,91 @@
+from collections.abc import Callable
+from typing import ContextManager
+
+from meiga import BoolResult, Error, Failure, Result, Success, isSuccess
+from meiga.decorators import meiga
+from petisco import Databases
+from petisco.base.application.patterns.crud_repository import CrudRepository
+from petisco.base.domain.errors.defaults.already_exists import (
+    AggregateAlreadyExistError,
+)
+from petisco.base.domain.errors.defaults.not_found import AggregateNotFoundError
+from petisco.base.domain.model.uuid import Uuid
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.src.task.shared.domain.task import Task
+from app.src.task.shared.infrastructure.sql.sql_task import SqlTask
+
+
+class SqlTaskRepository(CrudRepository[Task]):
+    session_scope: Callable[..., ContextManager[Session]]
+
+    def __init__(self):
+        self.session_scope = Databases.get_session_scope("sql-tasks")
+
+    @meiga
+    def save(self, task: Task) -> BoolResult:
+
+        with self.session_scope() as session:
+            query = select(SqlTask).where(
+                SqlTask.aggregate_id == task.aggregate_id.value
+            )
+            sql_task = session.execute(query).first()
+
+            if sql_task:
+                return Failure(AggregateAlreadyExistError(task.aggregate_id))
+
+            sql_task = SqlTask.from_domain(task)
+            session.add(sql_task)
+
+        return isSuccess
+
+    @meiga
+    def retrieve(self, aggregate_id: Uuid) -> Result[Task, Error]:
+        with self.session_scope() as session:
+            query = select(SqlTask).where(SqlTask.aggregate_id == aggregate_id.value)
+            sql_task = session.execute(query).first()
+
+            if sql_task:
+                return Failure(AggregateNotFoundError(aggregate_id))
+            task = sql_task.to_domain()
+
+            return Success(task)
+
+    def update(self, task: Task) -> BoolResult:
+        with self.session_scope() as session:
+            query = select(SqlTask).where(
+                SqlTask.aggregate_id == task.aggregate_id.value
+            )
+            sql_task = session.execute(query).first()
+
+            if sql_task:
+                return Failure(AggregateNotFoundError(task.aggregate_id))
+
+            sql_task = SqlTask.from_domain(task)
+            session.add(sql_task)
+
+        return isSuccess
+
+    def remove(self, aggregate_id: Uuid) -> BoolResult:
+        with self.session_scope() as session:
+            query = select(SqlTask).where(SqlTask.aggregate_id == aggregate_id.value)
+            sql_task = session.execute(query).first()
+
+            if sql_task:
+                return Failure(AggregateNotFoundError(aggregate_id))
+
+            session.remove(sql_task)
+
+        return isSuccess
+
+    def retrieve_all(self) -> Result[list[Task], Error]:
+        with self.session_scope() as session:
+            query = select(SqlTask)
+            sql_tasks = session.execute(query).all()
+            tasks = [sql_task.to_domain().values() for sql_task in sql_tasks]
+
+        return Success(tasks)
+
+    def clear(self):
+        Databases().remove("sql-tasks")

tests/modules/task/shared/integration/test_folder_task_crud_repository.py

@@ -3,19 +3,19 @@
 from petisco import AggregateAlreadyExistError, AggregateNotFoundError
 
 from app.src.task.shared.domain.task import Task
-from app.src.task.shared.infrastructure.folder_crud_repository import (
-    FolderTaskCrudRepository,
+from app.src.task.shared.infrastructure.folder_task_repository import (
+    FolderTaskRepository,
 )
 from tests.mothers.task_mother import TaskMother
 
 
 @pytest.mark.integration
 class TestFolderTaskCrudRepository:
-    repository: FolderTaskCrudRepository
+    repository: FolderTaskRepository
     task: Task
 
     def setup_method(self):
-        self.repository = FolderTaskCrudRepository("folder_task_database")
+        self.repository = FolderTaskRepository("folder_task_database")
         self.aggregate_root = TaskMother.any()
 
     def teardown_method(self):