Add task service for get task API (#12)

* Add task and label models

* Add tests for label and task model

* Remove json schema generator from PyObjectId

This is not required

* Add base mongo repository

* Add task repository

* Add label repository

* Add tests for base mongo repository

* Add tests for task repository

* Add tests for label repository

* Restructure health view and url

* Add a basic task view

* Add query param validator

* Add api error response

* Add exception handler for validation errors

* Add tests for exception handler

* Add tests for get task serializer

* Add get tasks api response, task and label DTOs

* Add task service

* Add tests for task service

* Use task service in task view

* Add tests for task view

* Resolve merge conflicts and move pagination settings to Django settings

* fix: resolve merge conflicts and fix syntax errors

- Fix merge conflicts across task views and fixtures
- Fix syntax error in test_task_service.py f-strings
- Update integration tests to use proper mocking
- Ensure all 67 tests pass successfully

* refactor: improve method signatures with named parameters

- Update TaskService.get_tasks to use named parameters with default values
- Refactor TaskView to pass parameters directly without intermediate variables
- Update tests to use named parameters in assertions

* refactor: implement Django paginator and improve error handling

- Replace custom pagination logic with Django's Paginator class
- Source pagination configuration values from Django settings
- Break down TaskService methods for better separation of concerns
- Add comprehensive error handling for validation and pagination
- Improve code organization and documentation
- Fix lint issues with unused exception variables
- Remove unused imports

* chore: remove comments from task service

* refactor: implement proper error handling and improve code quality

- Add error information to all exception responses
- Make createdBy a required field in DTOs
- Rename _build_page_url to build_page_url (made public)
- Remove unused exception variables
- Maintain 100% test coverage

---------

Co-authored-by: Achintya-Chatterjee <[email protected]>

Author: samarpan1738

todo/dto/label_dto.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+from pydantic import BaseModel
+
+from todo.dto.user_dto import UserDTO
+
+
+class LabelDTO(BaseModel):
+    name: str
+    color: str
+    createdAt: datetime | None = None
+    updatedAt: datetime | None = None
+    createdBy: UserDTO | None = None
+    updatedBy: UserDTO | None = None

todo/dto/responses/get_tasks_response.py

@@ -0,0 +1,8 @@
+from typing import List
+
+from todo.dto.responses.paginated_response import PaginatedResponse
+from todo.dto.task_dto import TaskDTO
+
+
+class GetTasksResponse(PaginatedResponse):
+    tasks: List[TaskDTO] = []

todo/dto/responses/paginated_response.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+from typing import Dict, Any, Optional
+
+
+class LinksData(BaseModel):
+    next: str | None = None
+    prev: str | None = None
+
+
+class PaginatedResponse(BaseModel):
+    links: LinksData | None = None
+    error: Optional[Dict[str, Any]] = None

todo/dto/task_dto.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+from typing import List
+from pydantic import BaseModel
+
+from todo.constants.task import TaskPriority, TaskStatus
+from todo.dto.label_dto import LabelDTO
+from todo.dto.user_dto import UserDTO
+
+
+class TaskDTO(BaseModel):
+    id: str
+    displayId: str
+    title: str
+    description: str | None = None
+    priority: TaskPriority | None = None
+    status: TaskStatus | None = None
+    assignee: UserDTO | None = None
+    isAcknowledged: bool | None = None
+    labels: List[LabelDTO] = []
+    startedAt: datetime | None = None
+    dueAt: datetime | None = None
+    createdAt: datetime
+    updatedAt: datetime | None = None
+    createdBy: UserDTO
+    updatedBy: UserDTO | None = None
+
+    class Config:
+        json_encoders = {TaskPriority: lambda x: x.name}

todo/dto/user_dto.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class UserDTO(BaseModel):
+    id: str
+    name: str

todo/repositories/task_repository.py

@@ -16,3 +16,15 @@ def list(cls, page: int, limit: int) -> List[TaskModel]:
     def count(cls) -> int:
         tasks_collection = cls.get_collection()
         return tasks_collection.count_documents({})
+
+    @classmethod
+    def get_all(cls) -> List[TaskModel]:
+        """
+        Get all tasks from the repository
+
+        Returns:
+            List[TaskModel]: List of all task models
+        """
+        tasks_collection = cls.get_collection()
+        tasks_cursor = tasks_collection.find()
+        return [TaskModel(**task) for task in tasks_cursor]

todo/services/__init__.py

@@ -0,0 +1 @@
+# Added this because without this file Django isn't able to auto detect the test files

todo/services/task_service.py

@@ -0,0 +1,143 @@
+from typing import List
+from dataclasses import dataclass
+from django.core.paginator import Paginator, EmptyPage
+from django.core.exceptions import ValidationError
+from django.urls import reverse_lazy
+from urllib.parse import urlencode
+
+from todo.dto.label_dto import LabelDTO
+from todo.dto.task_dto import TaskDTO
+from todo.dto.user_dto import UserDTO
+from todo.dto.responses.get_tasks_response import GetTasksResponse
+from todo.dto.responses.paginated_response import LinksData
+from todo.models.task import TaskModel
+from todo.repositories.task_repository import TaskRepository
+from todo.repositories.label_repository import LabelRepository
+from django.conf import settings
+
+
+@dataclass
+class PaginationConfig:
+    DEFAULT_PAGE: int = 1
+    DEFAULT_LIMIT: int = settings.REST_FRAMEWORK["DEFAULT_PAGINATION_SETTINGS"]["DEFAULT_PAGE_LIMIT"]
+    MAX_LIMIT: int = settings.REST_FRAMEWORK["DEFAULT_PAGINATION_SETTINGS"]["MAX_PAGE_LIMIT"]
+
+
+class TaskService:
+    @classmethod
+    def get_tasks(
+        cls, page: int = PaginationConfig.DEFAULT_PAGE, limit: int = PaginationConfig.DEFAULT_LIMIT
+    ) -> GetTasksResponse:
+        try:
+            cls._validate_pagination_params(page, limit)
+
+            tasks = TaskRepository.get_all()
+
+            if not tasks:
+                return GetTasksResponse(tasks=[], links=None)
+
+            paginator = Paginator(tasks, limit)
+
+            try:
+                current_page = paginator.page(page)
+
+                task_dtos = [cls.prepare_task_dto(task) for task in current_page.object_list]
+
+                links = cls._prepare_pagination_links(current_page=current_page, page=page, limit=limit)
+
+                return GetTasksResponse(tasks=task_dtos, links=links)
+
+            except EmptyPage:
+                return GetTasksResponse(
+                    tasks=[],
+                    links=None,
+                    error={"message": "Requested page exceeds available results", "code": "PAGE_NOT_FOUND"},
+                )
+
+        except ValidationError as e:
+            return GetTasksResponse(tasks=[], links=None, error={"message": str(e), "code": "VALIDATION_ERROR"})
+
+        except Exception:
+            return GetTasksResponse(
+                tasks=[], links=None, error={"message": "An unexpected error occurred", "code": "INTERNAL_ERROR"}
+            )
+
+    @classmethod
+    def _validate_pagination_params(cls, page: int, limit: int) -> None:
+        if page < 1:
+            raise ValidationError("Page must be a positive integer")
+
+        if limit < 1:
+            raise ValidationError("Limit must be a positive integer")
+
+        if limit > PaginationConfig.MAX_LIMIT:
+            raise ValidationError(f"Maximum limit of {PaginationConfig.MAX_LIMIT} exceeded")
+
+    @classmethod
+    def _prepare_pagination_links(cls, current_page, page: int, limit: int) -> LinksData:
+        next_link = None
+        prev_link = None
+
+        if current_page.has_next():
+            next_page = current_page.next_page_number()
+            next_link = cls.build_page_url(next_page, limit)
+
+        if current_page.has_previous():
+            prev_page = current_page.previous_page_number()
+            prev_link = cls.build_page_url(prev_page, limit)
+
+        return LinksData(next=next_link, prev=prev_link)
+
+    @classmethod
+    def build_page_url(cls, page: int, limit: int) -> str:
+        base_url = reverse_lazy("tasks")
+        query_params = urlencode({"page": page, "limit": limit})
+        return f"{base_url}?{query_params}"
+
+    @classmethod
+    def prepare_task_dto(cls, task_model: TaskModel) -> TaskDTO:
+        label_dtos = cls._prepare_label_dtos(task_model.labels) if task_model.labels else []
+
+        assignee = cls.prepare_user_dto(task_model.assignee) if task_model.assignee else None
+        created_by = cls.prepare_user_dto(task_model.createdBy)
+        updated_by = cls.prepare_user_dto(task_model.updatedBy) if task_model.updatedBy else None
+
+        return TaskDTO(
+            id=str(task_model.id),
+            displayId=task_model.displayId,
+            title=task_model.title,
+            description=task_model.description,
+            assignee=assignee,
+            isAcknowledged=task_model.isAcknowledged,
+            labels=label_dtos,
+            startedAt=task_model.startedAt,
+            dueAt=task_model.dueAt,
+            status=task_model.status,
+            priority=task_model.priority,
+            createdAt=task_model.createdAt,
+            updatedAt=task_model.updatedAt,
+            createdBy=created_by,
+            updatedBy=updated_by,
+        )
+
+    @classmethod
+    def _prepare_label_dtos(cls, label_ids: List[str]) -> List[LabelDTO]:
+        label_models = LabelRepository.list_by_ids(label_ids)
+
+        return [
+            LabelDTO(
+                name=label_model.name,
+                color=label_model.color,
+                createdAt=label_model.createdAt,
+                updatedAt=label_model.updatedAt if hasattr(label_model, "updatedAt") else None,
+                createdBy=cls.prepare_user_dto(label_model.createdBy),
+                updatedBy=cls.prepare_user_dto(label_model.updatedBy)
+                if hasattr(label_model, "updatedBy") and label_model.updatedBy
+                else None,
+            )
+            for label_model in label_models
+        ]
+
+    @classmethod
+    def prepare_user_dto(cls, user_id: str) -> UserDTO:
+        return UserDTO(id=user_id, name="SYSTEM")

todo/tests/fixtures/task.py

@@ -1,6 +1,7 @@
 from todo.constants.task import TaskPriority
 from todo.models.task import TaskModel
 from todo.constants.task import TaskStatus
+from todo.dto.task_dto import TaskDTO
 from bson import ObjectId
 
 tasks_db_data = [
@@ -37,3 +38,41 @@
 ]
 
 tasks_models = [TaskModel(**data) for data in tasks_db_data]
+
+
+task_dtos = [
+    TaskDTO(
+        id="672f7c5b775ee9f4471ff1dd",
+        displayId="#1",
+        title="created rest api",
+        priority=1,
+        status="TODO",
+        assignee={"id": "qMbT6M2GB65W7UHgJS4g", "name": "SYSTEM"},
+        isAcknowledged=False,
+        labels=[{"name": "Beginner Friendly", "color": "#fa1e4e"}],
+        isDeleted=False,
+        startedAt="2024-11-09T15:14:35.724000",
+        dueAt="2024-11-09T15:14:35.724000",
+        createdAt="2024-11-09T15:14:35.724000",
+        updatedAt="2024-10-18T15:55:14.802000Z",
+        createdBy={"id": "xQ1CkCncM8Novk252oAj", "name": "SYSTEM"},
+        updatedBy={"id": "Kn5N4Z3mdvpkv0HpqUCt", "name": "SYSTEM"},
+    ),
+    TaskDTO(
+        id="674c726ca89aab38040cb964",
+        displayId="#1",
+        title="task 2",
+        priority=1,
+        status="TODO",
+        assignee={"id": "qMbT6M2GB65W7UHgJS4g", "name": "SYSTEM"},
+        isAcknowledged=True,
+        labels=[{"name": "Beginner Friendly", "color": "#fa1e4e"}],
+        isDeleted=False,
+        startedAt="2024-11-09T15:14:35.724000",
+        dueAt="2024-11-09T15:14:35.724000",
+        createdAt="2024-11-09T15:14:35.724000",
+        updatedAt="2024-10-18T15:55:14.802000Z",
+        createdBy={"id": "xQ1CkCncM8Novk252oAj", "name": "SYSTEM"},
+        updatedBy={"id": "Kn5N4Z3mdvpkv0HpqUCt", "name": "SYSTEM"},
+    ),
+]

todo/tests/integration/test_tasks_pagination.py

@@ -1,8 +1,11 @@
 from unittest import TestCase
+from unittest.mock import patch
 from django.conf import settings
 from rest_framework.test import APIRequestFactory
 
 from todo.views.task import TaskView
+from todo.dto.responses.get_tasks_response import GetTasksResponse
+from todo.tests.fixtures.task import task_dtos
 
 
 class TaskPaginationIntegrationTest(TestCase):
@@ -12,20 +15,34 @@ def setUp(self):
         self.factory = APIRequestFactory()
         self.view = TaskView.as_view()
 
-    def test_pagination_settings_integration(self):
+    @patch("todo.services.task_service.TaskService.get_tasks")
+    def test_pagination_settings_integration(self, mock_get_tasks):
         """Test that the view and serializer correctly use Django settings for pagination"""
+        mock_get_tasks.return_value = GetTasksResponse(tasks=task_dtos)
+
         # Test with no query params (should use default limit)
         request = self.factory.get("/tasks")
         response = self.view(request)
-        
+
         # Check serializer validation passed and returned 200 OK
         self.assertEqual(response.status_code, 200)
-        
+
+        default_limit = settings.REST_FRAMEWORK["DEFAULT_PAGINATION_SETTINGS"]["DEFAULT_PAGE_LIMIT"]
+        mock_get_tasks.assert_called_with(page=1, limit=default_limit)
+
+        mock_get_tasks.reset_mock()
+
+        request = self.factory.get("/tasks", {"limit": "10"})
+        response = self.view(request)
+
+        self.assertEqual(response.status_code, 200)
+        mock_get_tasks.assert_called_with(page=1, limit=10)
+
         # Verify API rejects values above max limit
         max_limit = settings.REST_FRAMEWORK["DEFAULT_PAGINATION_SETTINGS"]["MAX_PAGE_LIMIT"]
         request = self.factory.get("/tasks", {"limit": str(max_limit + 1)})
         response = self.view(request)
-        
+
         # Should get a 400 error
         self.assertEqual(response.status_code, 400)
-        self.assertIn(str(max_limit), str(response.data)) 
+        self.assertIn(str(max_limit), str(response.data))