Adding storage to the Task MVP of TAD

pyproject.toml

@@ -30,6 +30,7 @@ jinja2 = "^3.1.4"
 pydantic-settings = "^2.2.1"
 psycopg2-binary = "^2.9.9"
 uvicorn = {extras = ["standard"], version = "^0.30.1"}
+pyyaml = "^6.0.1"
 
 
 [tool.poetry.group.test.dependencies]
@@ -66,6 +67,7 @@ line-ending = "lf"
 select = ["I", "SIM", "B", "UP", "F", "E", "S", "C90", "DTZ", "LOG", "PIE", "PT", "ERA", "W", "C", "TRY", "RUF"]
 fixable = ["ALL"]
 task-tags = ["TODO"]
+ignore = ["TRY003"]
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**.py" = ["S101"]

tad/core/config.py

@@ -85,7 +85,7 @@ def SQLALCHEMY_DATABASE_URI(self) -> str:
     @model_validator(mode="after")
     def _enforce_database_rules(self: SelfSettings) -> SelfSettings:
         if self.ENVIRONMENT != "local" and self.APP_DATABASE_SCHEME == "sqlite":
-            raise SettingsError("SQLite is not supported in production")  # noqa: TRY003
+            raise SettingsError("SQLite is not supported in production")
         return self
 
 

tad/migrations/env.py

@@ -57,7 +57,7 @@ def run_migrations_online():
     """
     configuration = config.get_section(config.config_ini_section)
     if configuration is None:
-        raise Exception("Failed to get configuration section")  # noqa: TRY003, TRY002
+        raise Exception("Failed to get configuration section")  # noqa: TRY002
     configuration["sqlalchemy.url"] = get_url()
     connectable = engine_from_config(
         configuration,

tad/models/system_card.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+from pydantic import Field as PydanticField  # type: ignore
+
+
+class SystemCard(BaseModel):
+    title: str = PydanticField(default=None)

tad/services/storage.py

@@ -0,0 +1,47 @@
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Any
+
+from yaml import dump
+
+
+class Writer(ABC):
+    @abstractmethod
+    def write(self, data: dict[str, Any]) -> None:
+        """This is an abstract method to write with the writer"""
+
+    @abstractmethod
+    def close(self) -> None:
+        """This is an abstract method to close the writer"""
+
+
+class WriterFactory:
+    @staticmethod
+    def get_writer(writer_type: str = "file", **kwargs: str) -> Writer:
+        match writer_type:
+            case "file":
+                if not all(k in kwargs for k in ("location", "filename")):
+                    raise KeyError("The `location` or `filename` variables are not provided as input for get_writer()")
+                return FileSystemWriteService(location=str(kwargs["location"]), filename=str(kwargs["filename"]))
+            case _:
+                raise ValueError(f"Unknown writer type: {writer_type}")
+
+
+class FileSystemWriteService(Writer):
+    def __init__(self, location: str = "./tests/data", filename: str = "system_card.yaml") -> None:
+        self.location = location
+        if not filename.endswith(".yaml"):
+            raise ValueError(f"Filename {filename} must end with .yaml instead of .{filename.split('.')[-1]}")
+        self.filename = filename
+
+    def write(self, data: dict[str, Any]) -> None:
+        if not Path(self.location).exists():
+            Path(self.location).mkdir()
+        with open(Path(self.location) / self.filename, "w") as f:
+            dump(data, f, default_flow_style=False, sort_keys=False)
+
+    def close(self):
+        """
+        This method is empty because with the `with` statement in the writer, Python will already close the writer
+        after usage.
+        """

tad/services/tasks.py

@@ -4,10 +4,12 @@
 
 from fastapi import Depends
 
+from tad.models.system_card import SystemCard
 from tad.models.task import Task
 from tad.models.user import User
 from tad.repositories.tasks import TasksRepository
 from tad.services.statuses import StatusesService
+from tad.services.storage import WriterFactory
 
 logger = logging.getLogger(__name__)
 
@@ -20,6 +22,10 @@ def __init__(
     ):
         self.repository = repository
         self.statuses_service = statuses_service
+        self.storage_writer = WriterFactory.get_writer(
+            writer_type="file", location="./output", filename="system_card.yaml"
+        )
+        self.system_card = SystemCard()
 
     def get_tasks(self, status_id: int) -> Sequence[Task]:
         return self.repository.find_by_status_id(status_id)
@@ -43,8 +49,8 @@ def move_task(
         task = self.repository.find_by_id(task_id)
 
         if status.name == "done":
-            # TODO implement logic for done
-            logging.warning("Task is done, we need to update a system card")
+            self.system_card.title = task.title
+            self.storage_writer.write(self.system_card.model_dump())
 
         # assign the task to the current user
         if status.name == "in_progress":

tests/core/test_exceptions.py

@@ -4,6 +4,6 @@
 
 def test_environment_settings():
     with pytest.raises(SettingsError) as exc_info:
-        raise SettingsError("Wrong settings")  # noqa: TRY003
+        raise SettingsError("Wrong settings")
 
     assert exc_info.value.message == "Wrong settings"

tests/services/test_storage.py

@@ -0,0 +1,75 @@
+from pathlib import Path
+
+import pytest
+from tad.models.system_card import SystemCard
+from tad.services.storage import WriterFactory
+from yaml import safe_load
+
+
+@pytest.fixture()
+def setup_and_teardown(tmp_path: Path) -> tuple[str, str]:
+    filename = "test.yaml"
+    return filename, str(tmp_path.absolute())
+
+
+def test_file_system_writer_empty_yaml(setup_and_teardown: tuple[str, str]) -> None:
+    filename, location = setup_and_teardown
+
+    storage_writer = WriterFactory.get_writer(writer_type="file", location=location, filename=filename)
+    storage_writer.write({})
+
+    assert Path.is_file(Path(location) / filename), True
+
+
+def test_file_system_writer_no_location_variable(setup_and_teardown: tuple[str, str]) -> None:
+    filename, _ = setup_and_teardown
+    with pytest.raises(
+        KeyError, match="The `location` or `filename` variables are not provided as input for get_writer()"
+    ):
+        WriterFactory.get_writer(writer_type="file", filename=filename)
+
+
+def test_file_system_writer_no_filename_variable(setup_and_teardown: tuple[str, str]) -> None:
+    _, location = setup_and_teardown
+    with pytest.raises(
+        KeyError, match="The `location` or `filename` variables are not provided as input for get_writer()"
+    ):
+        WriterFactory.get_writer(writer_type="file", location=location)
+
+
+def test_file_system_writer_yaml_with_content(setup_and_teardown: tuple[str, str]) -> None:
+    filename, location = setup_and_teardown
+    data = {"test": "test"}
+    storage_writer = WriterFactory.get_writer(writer_type="file", location=location, filename=filename)
+    storage_writer.write(data)
+
+    with open(Path(location) / filename) as f:
+        assert safe_load(f) == data, True
+
+
+def test_file_system_writer_with_system_card(setup_and_teardown: tuple[str, str]) -> None:
+    filename, location = setup_and_teardown
+    data = SystemCard()
+    data.title = "test"
+    data_dict = data.model_dump()
+
+    storage_writer = WriterFactory.get_writer(writer_type="file", location=location, filename=filename)
+    storage_writer.write(data_dict)
+
+    with open(Path(location) / filename) as f:
+        assert safe_load(f) == data_dict, True
+
+
+def test_abstract_writer_non_yaml_filename(setup_and_teardown: tuple[str, str]) -> None:
+    _, location = setup_and_teardown
+    filename = "test.csv"
+    with pytest.raises(
+        ValueError, match=f"Filename {filename} must end with .yaml instead of .{filename.split('.')[-1]}"
+    ):
+        WriterFactory.get_writer(writer_type="file", location=location, filename=filename)
+
+
+def test_not_valid_writer_type() -> None:
+    writer_type = "kafka"
+    with pytest.raises(ValueError, match=f"Unknown writer type: {writer_type}"):
+        WriterFactory.get_writer(writer_type=writer_type)

tests/services/test_system_cards.py

@@ -0,0 +1,23 @@
+import pytest
+from tad.models.system_card import SystemCard
+
+
+@pytest.fixture()
+def setup() -> SystemCard:
+    system_card = SystemCard()
+    return system_card
+
+
+def test_get_system_card(setup: SystemCard) -> None:
+    system_card = setup
+    expected = {"title": None}
+
+    assert system_card.model_dump() == expected
+
+
+def test_system_card_update(setup: SystemCard) -> None:
+    system_card = setup
+    expected = {"title": "IAMA 1.1"}
+    system_card.title = "IAMA 1.1"
+
+    assert system_card.model_dump() == expected