Add database models for get tasks API

.coveragerc

@@ -3,4 +3,5 @@
 omit =
     */__init__.py
     todo_project/wsgi.py
-    manage.py
+    manage.py
+    */tests/*

requirements.txt

@@ -1,3 +1,4 @@
+annotated-types==0.7.0
 asgiref==3.8.1
 cfgv==3.4.0
 coverage==7.6.4
@@ -9,9 +10,12 @@ filelock==3.16.1
 identify==2.6.1
 nodeenv==1.9.1
 platformdirs==4.3.6
+pydantic==2.10.1
+pydantic_core==2.27.1
 pymongo==4.10.1
 python-dotenv==1.0.1
 PyYAML==6.0.2
 ruff==0.7.1
 sqlparse==0.5.1
+typing_extensions==4.12.2
 virtualenv==20.27.0

todo/constants/task.py

@@ -0,0 +1,15 @@
+from enum import Enum
+
+
+class TaskStatus(Enum):
+    TODO = "TODO"
+    IN_PROGRESS = "IN_PROGRESS"
+    DEFERRED = "DEFERRED"
+    BLOCKED = "BLOCKED"
+    DONE = "DONE"
+
+
+class TaskPriority(Enum):
+    HIGH = 1
+    MEDIUM = 2
+    LOW = 3

todo/models/common/document.py

@@ -0,0 +1,20 @@
+from abc import ABC
+from bson import ObjectId
+
+from pydantic import BaseModel, Field
+
+from todo.models.common.pyobjectid import PyObjectId
+
+
+class Document(BaseModel, ABC):
+    id: PyObjectId | None = Field(None, alias="_id")
+
+    def __init_subclass__(cls, **kwargs):
+        super().__init_subclass__(**kwargs)
+        if not hasattr(cls, "collection_name") or not isinstance(cls.collection_name, str):
+            raise TypeError(f"Class {cls.__name__} must define a static `collection_name` field as a string.")
+
+    class Config:
+        from_attributes = True
+        json_encoders = {ObjectId: str}
+        populate_by_name = True

todo/models/common/pyobjectid.py

@@ -0,0 +1,15 @@
+from bson import ObjectId
+
+
+class PyObjectId(ObjectId):
+    @classmethod
+    def __get_validators__(cls):
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, value, field=None):
+        if value is None:
+            return None
+        if value is not None and ObjectId.is_valid(value):
+            return ObjectId(value)
+        raise ValueError(f"Invalid ObjectId: {value}")

todo/models/label.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from typing import ClassVar
+from todo.models.common.document import Document
+
+
+class LabelModel(Document):
+    collection_name: ClassVar[str] = "labels"
+
+    name: str
+    color: str
+    isDeleted: bool = False
+    createdAt: datetime
+    updatedAt: datetime | None = None
+    createdBy: str
+    updatedBy: str | None = None

todo/models/task.py

@@ -0,0 +1,42 @@
+from pydantic import BaseModel, Field
+from typing import ClassVar, List
+from datetime import datetime
+
+from todo.constants.task import TaskPriority, TaskStatus
+from todo.models.common.document import Document
+
+from todo.models.common.pyobjectid import PyObjectId
+from todo_project.db.config import DatabaseManager
+
+database_manager = DatabaseManager()
+
+
+class DeferredDetailsModel(BaseModel):
+    deferredAt: datetime | None = None
+    deferredTill: datetime | None = None
+    deferredBy: str | None = None
+
+    class Config:
+        from_attributes = True
+
+
+class TaskModel(Document):
+    collection_name: ClassVar[str] = "tasks"
+
+    id: PyObjectId | None = Field(None, alias="_id")
+    displayId: str
+    title: str
+    description: str | None = None
+    priority: TaskPriority | None = None
+    status: TaskStatus | None = None
+    assignee: str | None = None
+    isAcknowledged: bool | None = None
+    labels: List[PyObjectId] | None = []
+    isDeleted: bool = False
+    deferredDetails: DeferredDetailsModel | None = None
+    startedAt: datetime | None = None
+    dueAt: datetime | None = None
+    createdAt: datetime
+    updatedAt: datetime | None = None
+    createdBy: str
+    updatedBy: str | None = None

todo/tests/fixtures/label.py

@@ -0,0 +1,20 @@
+from bson import ObjectId
+from todo.models.label import LabelModel
+
+
+label_db_data = [
+    {
+        "_id": ObjectId("67478036eac9d93db7f59c35"),
+        "name": "Label 1",
+        "color": "#fa1e4e",
+        "createdAt": "2024-11-08T10:14:35",
+        "createdBy": "qMbT6M2GB65W7UHgJS4g",
+    },
+    {
+        "_id": ObjectId("67588c1ac2195684a575840c"),
+        "name": "Label 2",
+        "color": "#ea1e4e",
+        "createdAt": "2024-11-08T10:14:35",
+        "createdBy": "qMbT6M2GB65W7UHgJS4g",
+    },
+]

todo/tests/fixtures/task.py

@@ -0,0 +1,37 @@
+from todo.constants.task import TaskPriority
+from todo.models.task import TaskModel
+from todo.constants.task import TaskStatus
+from bson import ObjectId
+
+tasks_db_data = [
+    {
+        "id": ObjectId("672f7c5b775ee9f4471ff1dd"),
+        "displayId": "#1",
+        "title": "Task 1",
+        "description": "Test task 1",
+        "priority": TaskPriority.HIGH.value,
+        "status": TaskStatus.TODO.value,
+        "assignee": "qMbT6M2GB65W7UHgJS4g",
+        "isAcknowledged": True,
+        "labels": [ObjectId("67588c1ac2195684a575840c"), ObjectId("67478036eac9d93db7f59c35")],
+        "createdAt": "2024-11-08T10:14:35",
+        "updatedAt": "2024-11-08T15:14:35",
+        "createdBy": "qMbT6M2GB65W7UHgJS4g",
+        "updatedBy": "qMbT6M2GB65W7UHgJS4g",
+    },
+    {
+        "id": ObjectId("674c726ca89aab38040cb964"),
+        "displayId": "#2",
+        "title": "Task 2",
+        "description": "Test task 2",
+        "priority": TaskPriority.MEDIUM.value,
+        "status": TaskStatus.IN_PROGRESS.value,
+        "assignee": "qMbT6M2GB65W7UHgJS4g",
+        "isAcknowledged": False,
+        "labels": [ObjectId("67588c1ac2195684a575840c"), ObjectId("67478036eac9d93db7f59c35")],
+        "createdAt": "2024-11-08T10:14:35",
+        "updatedAt": "2024-11-08T15:14:35",
+        "createdBy": "qMbT6M2GB65W7UHgJS4g",
+        "updatedBy": "qMbT6M2GB65W7UHgJS4g",
+    },
+]

todo/tests/unit/models/__init__.py

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

todo/tests/unit/models/common/__init__.py

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

todo/tests/unit/models/common/test_document.py

@@ -0,0 +1,58 @@
+from typing import ClassVar
+from unittest import TestCase
+from bson import ObjectId
+from pydantic import Field
+from todo.models.common.document import Document
+
+
+class DocumentTests(TestCase):
+    def test_subclass_without_collection_name_raises_error(self):
+        with self.assertRaises(TypeError) as context:
+
+            class InvalidDocument(Document):
+                pass
+
+        self.assertIn("must define a static `collection_name` field as a string", str(context.exception))
+
+    def test_subclass_with_invalid_collection_name_type_raises_error(self):
+        with self.assertRaises(TypeError):
+
+            class InvalidDocument(Document):
+                collection_name: ClassVar[str] = 123
+
+    def test_subclass_with_valid_collection_name(self):
+        try:
+
+            class ValidDocument(Document):
+                collection_name: ClassVar[str] = "valid_collection"
+        except TypeError as e:
+            print(e)
+            self.fail("TypeError raised for valid Document subclass")
+
+    def test_id_field_alias(self):
+        class TestDocument(Document):
+            collection_name: ClassVar[str] = "test_collection"
+
+        obj_id = ObjectId()
+        doc = TestDocument.model_validate({"_id": obj_id})
+        self.assertEqual(doc.id, obj_id)
+        self.assertEqual(doc.model_dump(mode="json", by_alias=True)["_id"], str(obj_id))
+
+    def test_json_encoder_serializes_objectid(self):
+        class TestDocument(Document):
+            collection_name: ClassVar[str] = "test_collection"
+
+        obj_id = ObjectId()
+        doc = TestDocument(id=obj_id)
+        serialized = doc.model_dump_json()
+        self.assertIn(str(obj_id), serialized)
+
+    def test_populate_by_name_behavior(self):
+        class TestDocument(Document):
+            collection_name: ClassVar[str] = "test_collection"
+            field_one: str = Field(..., alias="fieldOne")
+
+        data = {"_id": ObjectId(), "fieldOne": "value"}
+        doc = TestDocument.model_validate(data)
+        self.assertEqual(doc.field_one, "value")
+        self.assertEqual(doc.model_dump(by_alias=True)["fieldOne"], "value")

todo/tests/unit/models/common/test_pyobjectid.py

@@ -0,0 +1,39 @@
+from unittest import TestCase
+from bson import ObjectId
+from pydantic import BaseModel, ValidationError
+
+from todo.models.common.pyobjectid import PyObjectId
+
+
+class PyObjectIdTests(TestCase):
+    def test_validate_valid_objectid(self):
+        valid_id = str(ObjectId())
+        validated = PyObjectId.validate(valid_id)
+        self.assertEqual(validated, ObjectId(valid_id))
+
+    def test_validate_invalid_objectid(self):
+        invalid_id = "invalid_objectid"
+        with self.assertRaises(ValueError) as context:
+            PyObjectId.validate(invalid_id)
+        self.assertIn(f"Invalid ObjectId: {invalid_id}", str(context.exception))
+
+    def test_validate_none(self):
+        self.assertIsNone(PyObjectId.validate(None))
+
+    def test_integration_with_pydantic_model(self):
+        class TestModel(BaseModel):
+            id: PyObjectId
+
+        valid_id = str(ObjectId())
+        instance = TestModel(id=valid_id)
+        self.assertEqual(instance.id, ObjectId(valid_id))
+
+        invalid_id = "invalid_objectid"
+        with self.assertRaises(ValidationError) as context:
+            TestModel(id=invalid_id)
+        self.assertIn(f"Invalid ObjectId: {invalid_id}", str(context.exception))
+
+        try:
+            TestModel(id=None)
+        except ValidationError:
+            self.fail("ValidationError raised for None id")

todo/tests/unit/models/test_label.py

@@ -0,0 +1,33 @@
+from unittest import TestCase
+from datetime import datetime
+
+from todo.models.label import LabelModel
+from todo.tests.fixtures.label import label_db_data
+from pydantic_core._pydantic_core import ValidationError
+
+
+class LabelModelTest(TestCase):
+    def setUp(self):
+        self.created_at = datetime.now()
+        self.valid_data = label_db_data[0]
+
+    def test_label_model_instantiates_with_valid_data(self):
+        label = LabelModel(**self.valid_data)
+        self.assertFalse(label.isDeleted)  # Default value
+        self.assertIsNone(label.updatedAt)  # Default value
+        self.assertIsNone(label.updatedBy)  # Default value
+
+    def test_lable_model_throws_error_when_missing_required_fields(self):
+        incomplete_data = self.valid_data.copy()
+        required_fields = ["name", "color", "createdAt", "createdBy"]
+        for field_name in required_fields:
+            del incomplete_data[field_name]
+        with self.assertRaises(ValidationError) as context:
+            LabelModel(**incomplete_data)
+
+        missing_fields_count = 0
+        for error in context.exception.errors():
+            self.assertEqual(error.get("type"), "missing")
+            self.assertIn(error.get("loc")[0], required_fields)
+            missing_fields_count += 1
+        self.assertEqual(missing_fields_count, len(required_fields))

todo/tests/unit/models/test_task.py

@@ -0,0 +1,47 @@
+from typing import List
+from unittest import TestCase
+
+from pydantic_core._pydantic_core import ValidationError
+from todo.models.task import TaskModel
+from todo.constants.task import TaskPriority, TaskStatus
+from todo.tests.fixtures.task import tasks_db_data
+
+
+class TaskModelTest(TestCase):
+    def setUp(self):
+        self.valid_task_data = tasks_db_data[0]
+
+    def test_task_model_instantiates_with_valid_data(self):
+        task = TaskModel(**self.valid_task_data)
+
+        self.assertEqual(task.priority, TaskPriority.HIGH)  # Enum value
+        self.assertEqual(task.status, TaskStatus.TODO)  # Enum value
+        self.assertFalse(task.isDeleted)  # Default value
+
+    def test_task_model_throws_error_when_missing_required_fields(self):
+        incomplete_data = self.valid_task_data.copy()
+        required_fields = ["displayId", "title", "createdAt", "createdBy"]
+        for field_name in required_fields:
+            del incomplete_data[field_name]
+
+        with self.assertRaises(ValidationError) as context:
+            TaskModel(**incomplete_data)
+
+        missing_fields_count = 0
+        for error in context.exception.errors():
+            self.assertEqual(error.get("type"), "missing")
+            self.assertIn(error.get("loc")[0], required_fields)
+            missing_fields_count += 1
+        self.assertEqual(missing_fields_count, len(required_fields))
+
+    def test_task_model_throws_error_when_invalid_enum_value(self):
+        invalid_data = self.valid_task_data.copy()
+        invalid_data["priority"] = "INVALID_PRIORITY"
+        invalid_data["status"] = "INVALID_STATUS"
+
+        with self.assertRaises(ValidationError) as context:
+            TaskModel(**invalid_data)
+        invalid_field_names = []
+        for error in context.exception.errors():
+            invalid_field_names.append(error.get("loc")[0])
+        self.assertEqual(invalid_field_names, ["priority", "status"])