refactor: map entity fields to request and response models

api/src/common/entity_mapper.py

@@ -0,0 +1,33 @@
+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]):
+        if name:
+            cls.__name__ = name
+
+        include_ = set(cls.__fields__.keys())
+
+        if include is not None:
+            include_ &= set(include)
+
+        exclude_ = set()
+        if exclude is not None:
+            exclude_ = set(exclude)
+        if include and exclude and set(include) & set(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,25 @@
-from dataclasses import asdict, dataclass, fields
+from __future__ import annotations
 
+from pydantic import BaseModel, Field
 
-@dataclass(frozen=True)
-class TodoItem:
+title_field = Field(
+    "title",
+    title="The title of the item",
+    max_length=30,
+    min_length=1,
+    example="Read about clean architecture",
+)
+
+
+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,7 @@
+from common.entity_mapper import filter_fields
+from entities.TodoItem import TodoItem
+
+
+@filter_fields(name="TodoItem", exclude=["user_id"])
+class TodoItemResponseModel(TodoItem):
+    pass

api/src/features/todo/todo_feature.py

@@ -11,30 +11,27 @@
     TodoRepositoryInterface,
 )
 
-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 AddTodoRequestModel, 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,
+    data: AddTodoRequestModel,
     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()
 
 
-@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 +41,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")
 @create_response(JSONResponse)
 def delete_todo_by_id(
     id: str,
@@ -54,15 +51,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,23 @@
 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
+from features.todo.shared_models import TodoItemResponseModel
 
 
-class AddTodoRequest(BaseModel):
-    title: str = Field(
-        ...,
-        title="The title of the item",
-        max_length=300,
-        min_length=1,
-        example="Read about clean architecture",
-    )
-
-
-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", include=["title"], exclude=[])
+class AddTodoRequestModel(TodoItem):
+    pass
 
 
 def add_todo_use_case(
-    data: AddTodoRequest,
+    data: AddTodoRequestModel,
     user_id: str,
     todo_repository: TodoRepositoryInterface,
-) -> AddTodoResponse:
+) -> TodoItemResponseModel:
     todo_item = TodoItem(id=str(uuid.uuid4()), title=data.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,17 @@
 )
 
 
-class DeleteTodoByIdResponse(BaseModel):
+class DeleteTodoByIdResponseModel(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
+) -> DeleteTodoByIdResponseModel:
     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 DeleteTodoByIdResponseModel(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,14 @@
 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(cast(TodoItemResponseModel, todo_item))

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

@@ -1,37 +1,26 @@
-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", include=["title", "is_completed"], exclude=[])
+class UpdateTodoRequest(TodoItem):
+    pass
 
 
 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/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 AddTodoRequestModel, add_todo_use_case
 
 
 def test_add_with_valid_title_should_return_todo(todo_repository: TodoRepositoryInterface):
-    data = AddTodoRequest(title="new todo")
+    data = AddTodoRequestModel(title="new todo")
     result = add_todo_use_case(data, user_id="xyz", todo_repository=todo_repository)
     assert result.title == data.title
 
 
 def test_add_with_empty_title_should_throw_validation_error(todo_repository: TodoRepositoryInterface):
     with pytest.raises(ValidationError):
-        data = AddTodoRequest(title="")
+        data = AddTodoRequestModel(title="")
         add_todo_use_case(data, user_id="xyz", todo_repository=todo_repository)

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

@@ -5,14 +5,14 @@
     TodoRepositoryInterface,
 )
 from features.todo.use_cases.delete_todo_by_id import (
-    DeleteTodoByIdResponse,
+    DeleteTodoByIdResponseModel,
     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: DeleteTodoByIdResponseModel = delete_todo_use_case(id=id, user_id="xyz", todo_repository=todo_repository)
     assert result.success
 
 

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

@@ -6,15 +6,12 @@
 from data_providers.repository_interfaces.TodoRepositoryInterface import (
     TodoRepositoryInterface,
 )
-from features.todo.use_cases.get_todo_by_id import (
-    GetTodoByIdResponse,
-    get_todo_by_id_use_case,
-)
+from features.todo.use_cases.get_todo_by_id import get_todo_by_id_use_case
 
 
 def test_get_todo_by_id_should_return_todo(todo_repository: TodoRepositoryInterface, todo_test_data: Dict[str, dict]):
     id = "dh2109"
-    todo: GetTodoByIdResponse = get_todo_by_id_use_case(id, user_id="xyz", todo_repository=todo_repository)
+    todo = get_todo_by_id_use_case(id, user_id="xyz", todo_repository=todo_repository)
     assert todo.title == todo_test_data[id]["title"]
 
 

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

@@ -8,4 +8,4 @@ def test_update_todo_should_return_success(todo_repository: TodoRepositoryInterf
     id = "dh2109"
     data = UpdateTodoRequest(title="new title", is_completed=False)
     result = update_todo_use_case(id, data, user_id="xyz", todo_repository=todo_repository)
-    assert result.success
+    assert data.title == result.title