Add task service for get task API

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,10 @@
+from pydantic import BaseModel
+
+
+class LinksData(BaseModel):
+    next: str | None = None
+    prev: str | None = None
+
+
+class PaginatedResponse(BaseModel):
+    links: LinksData | None = 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/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,65 @@
+from django.urls import reverse_lazy
+from todo.dto.label_dto import LabelDTO
+from todo.dto.responses.get_tasks_response import GetTasksResponse
+from todo.dto.responses.paginated_response import LinksData
+from todo.dto.task_dto import TaskDTO
+from todo.dto.user_dto import UserDTO
+from todo.models.task import TaskModel
+from todo.repositories.label_repository import LabelRepository
+from todo.repositories.task_repository import TaskRepository
+
+
+class TaskService:
+    tasks_api_base_url = reverse_lazy("tasks")
+
+    @classmethod
+    def get_tasks(cls, page, limit) -> GetTasksResponse:
+        response = GetTasksResponse()
+        tasks_count = TaskRepository.count()
+        tasks_skip_count = (page - 1) * limit
+        if tasks_count <= tasks_skip_count:
+            return response
+
+        tasks = TaskRepository.list(page, limit)
+        task_dicts = list(map(cls.prepare_task_dto, tasks))
+        response.tasks = task_dicts
+        links_data = LinksData()
+        if page > 1:
+            links_data.prev = f"{cls.tasks_api_base_url}?page={page-1}&limit={limit}"
+
+        if tasks_count > tasks_skip_count + limit:
+            links_data.next = f"{cls.tasks_api_base_url}?page={page+1}&limit={limit}"
+
+        if links_data.prev is not None or links_data.next is not None:
+            response.links = links_data
+        return response
+
+    @classmethod
+    def prepare_task_dto(cls, task: TaskModel) -> TaskDTO:
+        task_dict = task.model_dump(mode="json", exclude={"description", "isAcknowledged"})
+        if len(task.labels) > 0:
+            task_dict["labels"] = [
+                LabelDTO(
+                    **{
+                        "name": label.name,
+                        "color": label.color,
+                    }
+                )
+                for label in LabelRepository.list_by_ids(task.labels)
+            ]
+        task_dict["createdBy"] = cls.prepare_user_dto(task.createdBy)
+        if task.assignee is not None:
+            task_dict["assignee"] = cls.prepare_user_dto(task.assignee)
+        if task.updatedBy is not None:
+            task_dict["updatedBy"] = cls.prepare_user_dto(task.updatedBy)
+        return TaskDTO(**task_dict)
+
+    @classmethod
+    def prepare_user_dto(cls, user_id: str) -> UserDTO:
+        UserDTO.model_fields
+        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/unit/services/__init__.py

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

todo/tests/unit/services/test_task_service.py

@@ -0,0 +1,77 @@
+from unittest.mock import Mock, patch
+from unittest import TestCase
+
+from todo.dto.responses.get_tasks_response import GetTasksResponse
+from todo.dto.responses.paginated_response import LinksData
+from todo.dto.user_dto import UserDTO
+from todo.services.task_service import TaskService
+from todo.dto.task_dto import TaskDTO
+from todo.tests.fixtures.task import tasks_models
+from todo.tests.fixtures.label import label_models
+
+
+class TaskServiceTests(TestCase):
+    @patch("todo.services.task_service.reverse_lazy", return_value="/v1/tasks")
+    def setUp(self, mock_reverse_lazy):
+        self.mock_reverse_lazy = mock_reverse_lazy
+
+    @patch("todo.services.task_service.TaskRepository.count")
+    @patch("todo.services.task_service.TaskRepository.list")
+    @patch("todo.services.task_service.LabelRepository.list_by_ids")
+    def test_get_tasks_returns_paginated_response(
+        self, mock_label_repo: Mock, mock_task_repo: Mock, mock_task_count: Mock
+    ):
+        mock_task_repo.return_value = [tasks_models[0]]
+        mock_label_repo.return_value = label_models
+        mock_task_count.return_value = 5
+
+        response: GetTasksResponse = TaskService.get_tasks(2, 1)
+        self.assertIsInstance(response, GetTasksResponse)
+        self.assertEqual(len(response.tasks), 1)
+
+        self.assertIsInstance(response.links, LinksData)
+        self.assertEqual(response.links.next, f"{self.mock_reverse_lazy("tasks")}?page=3&limit=1")
+        self.assertEqual(response.links.prev, f"{self.mock_reverse_lazy("tasks")}?page=1&limit=1")
+
+    @patch("todo.services.task_service.TaskRepository.count")
+    @patch("todo.services.task_service.TaskRepository.list")
+    @patch("todo.services.task_service.LabelRepository.list_by_ids")
+    def test_get_tasks_doesnt_returns_prev_link_for_first_page(
+        self, mock_label_repo: Mock, mock_task_repo: Mock, mock_task_count: Mock
+    ):
+        mock_task_repo.return_value = [tasks_models[0]]
+        mock_label_repo.return_value = label_models
+        mock_task_count.return_value = 10
+
+        response: GetTasksResponse = TaskService.get_tasks(1, 1)
+
+        self.assertIsNone(response.links.prev)
+
+    @patch("todo.services.task_service.TaskRepository.count")
+    def test_get_tasks_returns_zero_tasks_if_no_tasks_present(self, mock_task_count: Mock):
+        mock_task_count.return_value = 5
+
+        response: GetTasksResponse = TaskService.get_tasks(2, 10)
+        self.assertIsInstance(response, GetTasksResponse)
+        self.assertEqual(len(response.tasks), 0)
+        self.assertIsNone(response.links)
+
+    @patch("todo.services.task_service.LabelRepository.list_by_ids")
+    def test_prepare_task_dto_maps_model_to_dto(self, mock_label_repo: Mock):
+        task_model = tasks_models[0]
+        mock_label_repo.return_value = label_models
+
+        result: TaskDTO = TaskService.prepare_task_dto(task_model)
+
+        mock_label_repo.assert_called_once_with(task_model.labels)
+
+        self.assertIsInstance(result, TaskDTO)
+        self.assertEqual(result.id, str(task_model.id))
+
+    def test_prepare_user_dto_maps_model_to_dto(self):
+        user_id = tasks_models[0].assignee
+        result: UserDTO = TaskService.prepare_user_dto(user_id)
+
+        self.assertIsInstance(result, UserDTO)
+        self.assertEqual(result.id, user_id)
+        self.assertEqual(result.name, "SYSTEM")

todo/tests/unit/views/test_task.py

@@ -0,0 +1,60 @@
+from rest_framework.test import APISimpleTestCase, APIClient
+from rest_framework.reverse import reverse
+from rest_framework import status
+from unittest.mock import patch, Mock
+from rest_framework.response import Response
+from todo.constants.task import DEFAULT_PAGE_LIMIT
+from todo.dto.responses.get_tasks_response import GetTasksResponse
+from todo.tests.fixtures.task import task_dtos
+
+
+class TaskViewTests(APISimpleTestCase):
+    def setUp(self):
+        self.client = APIClient()
+        self.url = reverse("tasks")
+        self.valid_params = {"page": 1, "limit": 10}
+
+    @patch("todo.services.task_service.TaskService.get_tasks")
+    def test_get_tasks_returns_200_for_valid_params(self, mock_get_tasks: Mock):
+        mock_get_tasks.return_value = GetTasksResponse(tasks=task_dtos)
+
+        response: Response = self.client.get(self.url, self.valid_params)
+
+        mock_get_tasks.assert_called_once_with(1, 10)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        expected_response = mock_get_tasks.return_value.model_dump(mode="json", exclude_none=True)
+        self.assertDictEqual(response.data, expected_response)
+
+    @patch("todo.services.task_service.TaskService.get_tasks")
+    def test_get_tasks_returns_200_without_params(self, mock_get_tasks: Mock):
+        mock_get_tasks.return_value = GetTasksResponse(tasks=task_dtos)
+
+        response: Response = self.client.get(self.url)
+        mock_get_tasks.assert_called_once_with(1, DEFAULT_PAGE_LIMIT)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+    def test_get_tasks_returns_400_for_invalid_query_params(self):
+        invalid_params = {
+            "page": "invalid",
+            "limit": -1,
+        }
+
+        response: Response = self.client.get(self.url, invalid_params)
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        expected_response = {
+            "statusCode": 400,
+            "message": "Invalid request",
+            "errors": [
+                {"source": {"parameter": "page"}, "detail": "A valid integer is required."},
+                {"source": {"parameter": "limit"}, "detail": "limit must be greater than or equal to 1"},
+            ],
+        }
+        response_data = response.data
+
+        self.assertEqual(response_data["statusCode"], expected_response["statusCode"])
+        self.assertEqual(response_data["message"], expected_response["message"], "Error message mismatch")
+
+        for actual_error, expected_error in zip(response_data["errors"], expected_response["errors"]):
+            self.assertEqual(actual_error["source"]["parameter"], expected_error["source"]["parameter"])
+            self.assertEqual(actual_error["detail"], expected_error["detail"])

todo/views/task.py

@@ -3,10 +3,14 @@
 from rest_framework import status
 from rest_framework.request import Request
 from todo.serializers.get_tasks_serializer import GetTaskQueryParamsSerializer
+from todo.services.task_service import TaskService
 
 
 class TaskView(APIView):
     def get(self, request: Request):
         query = GetTaskQueryParamsSerializer(data=request.query_params)
         query.is_valid(raise_exception=True)
-        return Response({}, status.HTTP_200_OK)
+        page = query.validated_data["page"]
+        limit = query.validated_data["limit"]
+        response = TaskService.get_tasks(page, limit)
+        return Response(response.model_dump(mode="json", exclude_none=True), status.HTTP_200_OK)