refactor: map entity fields to request and response models

api/src/common/entity_mapper.py

@@ -0,0 +1,37 @@
+from typing import Optional, Type
+
+from pydantic import BaseModel
+
+
+def filter_fields(
+    name: Optional[str] = None,
+    include: Optional[list[str]] = None,
+    exclude: Optional[list[str]] = None,
+):
+    """Return a decorator to filter model fields"""
+
+    def decorator(cls: Type[BaseModel]):
+        config = cls.Config()
+        to_include = getattr(config, "include", include)
+        to_exclude = getattr(config, "exclude", exclude)
+
+        if name:
+            cls.__name__ = name
+
+        include_ = set(cls.__fields__.keys())
+
+        if to_include is not None:
+            include_ &= set(to_include)
+
+        exclude_ = set()
+        if to_exclude is not None:
+            exclude_ = set(to_exclude)
+        if to_include and to_exclude and set(to_include) & set(to_exclude):
+            raise ValueError("include and exclude cannot contain the same fields")
+
+        for field in list(cls.__fields__):
+            if field not in include_ or field in exclude_:
+                del cls.__fields__[field]
+        return cls
+
+    return decorator

api/src/data_providers/repositories/TodoRepository.py

@@ -9,7 +9,7 @@
 
 
 def to_dict(todo_item: TodoItem):
-    dict = todo_item.__dict__
+    dict = todo_item.dict()
     dict["_id"] = todo_item.id
     return dict
 

api/src/entities/TodoItem.py

@@ -1,19 +1,34 @@
-from dataclasses import asdict, dataclass, fields
+from __future__ import annotations
 
+from typing import List
 
-@dataclass(frozen=True)
-class TodoItem:
+from pydantic import BaseModel, Field
+
+title_field = Field(
+    ..., title="The title of the item", max_length=30, min_length=1, example="Read about clean architecture"
+)
+
+
+class User(BaseModel):
+    id: str
+    todos: List[TodoList]
+
+
+class TodoList(BaseModel):
+    id: str
+    user: User
+    items: List[TodoItem]
+
+
+class TodoItem(BaseModel):
     id: str
     user_id: str
-    title: str
+    title: str = title_field
     is_completed: bool = False
 
-    def to_dict(self):
-        return asdict(self)
-
     @classmethod
     def from_dict(cls, dict_) -> "TodoItem":
-        class_fields = {f.name for f in fields(cls)}
+        class_fields = {field for field in cls.__fields__}
         if "_id" in dict_:
             dict_["id"] = dict_.pop("_id")
         data = {k: v for k, v in dict_.items() if k in class_fields}

api/src/features/todo/shared_models.py

@@ -0,0 +1,8 @@
+from common.entity_mapper import filter_fields
+from entities.TodoItem import TodoItem
+
+
+@filter_fields(name="TodoItem")
+class TodoItemResponseModel(TodoItem):
+    class Config:
+        exclude = ["user_id"]

api/src/features/todo/todo_feature.py

@@ -10,31 +10,29 @@
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
+from features.todo.use_cases.delete_todo_by_id import DeleteTodoResponse
 
-from .use_cases.add_todo import AddTodoRequest, AddTodoResponse, add_todo_use_case
-from .use_cases.delete_todo_by_id import DeleteTodoByIdResponse, delete_todo_use_case
-from .use_cases.get_todo_all import GetTodoAllResponse, get_todo_all_use_case
-from .use_cases.get_todo_by_id import GetTodoByIdResponse, get_todo_by_id_use_case
-from .use_cases.update_todo import (
-    UpdateTodoRequest,
-    UpdateTodoResponse,
-    update_todo_use_case,
-)
+from .shared_models import TodoItemResponseModel
+from .use_cases.add_todo import add_todo_use_case
+from .use_cases.delete_todo_by_id import delete_todo_use_case
+from .use_cases.get_todo_all import get_todo_all_use_case
+from .use_cases.get_todo_by_id import get_todo_by_id_use_case
+from .use_cases.update_todo import UpdateTodoRequest, update_todo_use_case
 
 router = APIRouter(tags=["todos"], prefix="/todos")
 
 
-@router.post("", operation_id="create", response_model=AddTodoResponse)
+@router.post("", operation_id="create", response_model=TodoItemResponseModel)
 @create_response(JSONResponse)
 def add_todo(
-    data: AddTodoRequest,
+    title: str,
     user: User = Depends(auth_with_jwt),
     todo_repository: TodoRepositoryInterface = Depends(get_todo_repository),
 ):
-    return add_todo_use_case(data=data, user_id=user.user_id, todo_repository=todo_repository).dict()
+    return add_todo_use_case(title=title, user_id=user.user_id, todo_repository=todo_repository).dict()
 
 
-@router.get("/{id}", operation_id="get_by_id", response_model=GetTodoByIdResponse)
+@router.get("/{id}", operation_id="get_by_id", response_model=TodoItemResponseModel)
 @create_response(JSONResponse)
 def get_todo_by_id(
     id: str,
@@ -44,7 +42,7 @@ def get_todo_by_id(
     return get_todo_by_id_use_case(id=id, user_id=user.user_id, todo_repository=todo_repository).dict()
 
 
-@router.delete("/{id}", operation_id="delete_by_id", response_model=DeleteTodoByIdResponse)
+@router.delete("/{id}", operation_id="delete_by_id", response_model=DeleteTodoResponse)
 @create_response(JSONResponse)
 def delete_todo_by_id(
     id: str,
@@ -54,15 +52,15 @@ def delete_todo_by_id(
     return delete_todo_use_case(id=id, user_id=user.user_id, todo_repository=todo_repository).dict()
 
 
-@router.get("", operation_id="get_all", response_model=List[GetTodoAllResponse])
+@router.get("", operation_id="get_all", response_model=List[TodoItemResponseModel])
 @create_response(JSONResponse)
 def get_todo_all(
     user: User = Depends(auth_with_jwt), todo_repository: TodoRepositoryInterface = Depends(get_todo_repository)
 ):
     return [todo.dict() for todo in get_todo_all_use_case(user_id=user.user_id, todo_repository=todo_repository)]
 
 
-@router.put("/{id}", operation_id="update_by_id", response_model=UpdateTodoResponse)
+@router.put("/{id}", operation_id="update_by_id", response_model=TodoItemResponseModel)
 @create_response(JSONResponse)
 def update_todo(
     id: str,

api/src/features/todo/use_cases/add_todo.py

@@ -1,38 +1,24 @@
 import uuid
 
-from pydantic import BaseModel, Field
-
+from common.entity_mapper import filter_fields
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from entities.TodoItem import TodoItem
-
-
-class AddTodoRequest(BaseModel):
-    title: str = Field(
-        ...,
-        title="The title of the item",
-        max_length=300,
-        min_length=1,
-        example="Read about clean architecture",
-    )
-
+from entities.TodoItem import TodoItem, title_field
+from features.todo.shared_models import TodoItemResponseModel
 
-class AddTodoResponse(BaseModel):
-    id: str = Field(example="vytxeTZskVKR7C7WgdSP3d")
-    title: str = Field(example="Read about clean architecture")
-    is_completed: bool = False
 
-    @staticmethod
-    def from_entity(todo_item: TodoItem) -> "AddTodoResponse":
-        return AddTodoResponse(id=todo_item.id, title=todo_item.title, is_completed=todo_item.is_completed)
+@filter_fields(name="AddTodo")
+class AddTodoRequestModel(TodoItem):
+    class Config:
+        include = ["title"]
 
 
 def add_todo_use_case(
-    data: AddTodoRequest,
     user_id: str,
     todo_repository: TodoRepositoryInterface,
-) -> AddTodoResponse:
-    todo_item = TodoItem(id=str(uuid.uuid4()), title=data.title, user_id=user_id)
+    title: str = title_field,
+) -> TodoItemResponseModel:
+    todo_item = TodoItem(id=str(uuid.uuid4()), title=title, user_id=user_id)
     todo_repository.create(todo_item)
-    return AddTodoResponse.from_entity(todo_item)
+    return TodoItemResponseModel.parse_obj(todo_item)

api/src/features/todo/use_cases/delete_todo_by_id.py

@@ -6,15 +6,15 @@
 )
 
 
-class DeleteTodoByIdResponse(BaseModel):
+class DeleteTodoResponse(BaseModel):
     success: bool = Field(...)
 
 
-def delete_todo_use_case(id: str, user_id: str, todo_repository: TodoRepositoryInterface) -> DeleteTodoByIdResponse:
+def delete_todo_use_case(id: str, user_id: str, todo_repository: TodoRepositoryInterface) -> DeleteTodoResponse:
     todo_item = todo_repository.get(id)
     if todo_item is None:
         raise NotFoundException
     if todo_item.user_id != user_id:
         raise MissingPrivilegeException
     todo_repository.delete(id)
-    return DeleteTodoByIdResponse(success=True)
+    return DeleteTodoResponse(success=True)

api/src/features/todo/use_cases/get_todo_all.py

@@ -1,29 +1,17 @@
 from typing import List
 
-from pydantic import BaseModel, Field
-
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from entities.TodoItem import TodoItem
-
-
-class GetTodoAllResponse(BaseModel):
-    id: str = Field(...)
-    title: str = Field(...)
-    is_completed: bool
-
-    @staticmethod
-    def from_entity(todo_item: TodoItem):
-        return GetTodoAllResponse(id=todo_item.id, title=todo_item.title, is_completed=todo_item.is_completed)
+from features.todo.shared_models import TodoItemResponseModel
 
 
 def get_todo_all_use_case(
     user_id: str,
     todo_repository: TodoRepositoryInterface,
-) -> List[GetTodoAllResponse]:
+) -> List[TodoItemResponseModel]:
     return [
-        GetTodoAllResponse.from_entity(todo_item)
+        TodoItemResponseModel.parse_obj(todo_item)
         for todo_item in todo_repository.get_all()
         if todo_item.user_id == user_id
     ]

api/src/features/todo/use_cases/get_todo_by_id.py

@@ -1,26 +1,12 @@
-from typing import cast
-
-from pydantic import BaseModel, Field
-
 from common.exceptions import MissingPrivilegeException
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from entities.TodoItem import TodoItem
-
-
-class GetTodoByIdResponse(BaseModel):
-    id: str = Field(...)
-    title: str = Field(...)
-    is_completed: bool = False
-
-    @staticmethod
-    def from_entity(todo_item: TodoItem) -> "GetTodoByIdResponse":
-        return GetTodoByIdResponse(id=todo_item.id, title=todo_item.title, is_completed=todo_item.is_completed)
+from features.todo.shared_models import TodoItemResponseModel
 
 
-def get_todo_by_id_use_case(id: str, user_id: str, todo_repository: TodoRepositoryInterface) -> GetTodoByIdResponse:
+def get_todo_by_id_use_case(id: str, user_id: str, todo_repository: TodoRepositoryInterface) -> TodoItemResponseModel:
     todo_item = todo_repository.get(id)
     if todo_item.user_id != user_id:
         raise MissingPrivilegeException
-    return GetTodoByIdResponse.from_entity(cast(TodoItem, todo_item))
+    return TodoItemResponseModel.parse_obj(todo_item)

api/src/features/todo/use_cases/update_todo.py

@@ -1,37 +1,27 @@
-from pydantic import BaseModel, Field
-
+from common.entity_mapper import filter_fields
 from common.exceptions import MissingPrivilegeException
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
 from entities.TodoItem import TodoItem
+from features.todo.shared_models import TodoItemResponseModel
 
 
-class UpdateTodoRequest(BaseModel):
-    title: str = Field(
-        "",
-        title="The title of the item",
-        max_length=300,
-        min_length=1,
-    )
-    is_completed: bool
-
-
-class UpdateTodoResponse(BaseModel):
-    success: bool = Field(...)
+@filter_fields(name="UpdateTodo")
+class UpdateTodoRequest(TodoItem):
+    class Config:
+        include = ["title", "is_completed"]
 
 
 def update_todo_use_case(
     id: str,
     data: UpdateTodoRequest,
     user_id: str,
     todo_repository: TodoRepositoryInterface,
-) -> UpdateTodoResponse:
+) -> TodoItemResponseModel:
     todo_item = todo_repository.get(id)
     if todo_item.user_id != user_id:
         raise MissingPrivilegeException
-
     updated_todo_item = TodoItem(id=todo_item.id, title=data.title, is_completed=data.is_completed, user_id=user_id)
-    if todo_repository.update(updated_todo_item):
-        return UpdateTodoResponse(success=True)
-    return UpdateTodoResponse(success=False)
+    todo_repository.update(updated_todo_item)
+    return TodoItemResponseModel.parse_obj(updated_todo_item)

api/src/tests/integration/features/todo/test_todo_feature.py

@@ -59,7 +59,7 @@ def test_update_todo(self, test_app):
         response = test_app.put("/todos/1", json={"title": "title 1 updated", "is_completed": False})
 
         assert response.status_code == HTTP_200_OK
-        assert response.json()["success"]
+        assert response.json()["title"] == "title 1 updated"
 
     def test_update_todo_should_return_not_found(self, test_app):
         response = test_app.put("/todos/unknown", json={"title": "something", "is_completed": False})

api/src/tests/unit/features/todo/use_cases/test_add_todo.py

@@ -4,16 +4,16 @@
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from features.todo.use_cases.add_todo import AddTodoRequest, add_todo_use_case
+from features.todo.use_cases.add_todo import add_todo_use_case
 
 
 def test_add_with_valid_title_should_return_todo(todo_repository: TodoRepositoryInterface):
-    data = AddTodoRequest(title="new todo")
-    result = add_todo_use_case(data, user_id="xyz", todo_repository=todo_repository)
-    assert result.title == data.title
+    title = "new todo"
+    result = add_todo_use_case(title=title, user_id="xyz", todo_repository=todo_repository)
+    assert result.title == title
 
 
 def test_add_with_empty_title_should_throw_validation_error(todo_repository: TodoRepositoryInterface):
     with pytest.raises(ValidationError):
-        data = AddTodoRequest(title="")
-        add_todo_use_case(data, user_id="xyz", todo_repository=todo_repository)
+        title = ""
+        add_todo_use_case(title=title, user_id="xyz", todo_repository=todo_repository)

api/src/tests/unit/features/todo/use_cases/test_delete_todo_by_id.py

@@ -4,15 +4,12 @@
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from features.todo.use_cases.delete_todo_by_id import (
-    DeleteTodoByIdResponse,
-    delete_todo_use_case,
-)
+from features.todo.use_cases.delete_todo_by_id import delete_todo_use_case
 
 
 def test_delete_todo_should_return_success(todo_repository: TodoRepositoryInterface):
     id = "dh2109"
-    result: DeleteTodoByIdResponse = delete_todo_use_case(id=id, user_id="xyz", todo_repository=todo_repository)
+    result = delete_todo_use_case(id=id, user_id="xyz", todo_repository=todo_repository)
     assert result.success