feat: optimized models, views and caching, added Signals for task_app

task_app/apps.py

@@ -4,3 +4,6 @@ from django.apps import AppConfig
 class TaskAppConfig(AppConfig):
     default_auto_field = 'django.db.models.BigAutoField'
     name = 'task_app'
+
+    def ready(self):
+        import task_app.signals

task_app/caching.py

@@ -4,18 +4,11 @@ from .models import Task
 
 CACHE_TIMEOUT = 3600
 
-def get_cached_tasks():
-    return _get_or_set_cache("all_tasks", Task.objects.all())
-
-def get_cached_tasks_by_status(status):
-    return _get_or_set_cache(f"tasks_by_status_{status}", Task.objects.filter(status=status))
-
 def get_cached_task_by_id(task_id):
     return _get_or_set_cache(f"task_{task_id}", Task.objects.filter(pk=task_id), single=True)
 
 def _get_or_set_cache(cache_key, queryset, single=False):
     cached_json = cache.get(cache_key)
-    
     if cached_json:
         try:
             deserialized = list(serializers.deserialize("json", cached_json))

task_app/models.py

@@ -39,29 +39,4 @@ class AssignedTask(models.Model):
     task = models.ForeignKey(Task, related_name="assigned_tasks", on_delete=models.CASCADE)
     
     def __str__(self):
-        return f"{self.user_id} - {self.task}"
+        return f"{self.user_id} - {self.task}"
-
-# Cache invalidation
-@receiver(post_save, sender=Task)
-@receiver(post_delete, sender=Task)
-def clear_task_cache(sender, instance, **kwargs):
-    """Invalidate cache for the affected task status and individual task"""
-    cache.delete("all_tasks")
-
-    cache_key = f"tasks_by_status_{instance.status}"
-    cache.delete(cache_key)
-
-    cache_key_task = f"task_{instance.id}"
-    cache.delete(cache_key_task)
-
-@receiver(post_save, sender=AssignedTask)
-@receiver(post_delete, sender=AssignedTask)
-def clear_assigned_task_cache(sender, instance, **kwargs):
-    """Invalidate cache when a task assignment changes"""
-    cache_key = f"tasks_by_status_{instance.task.status}"
-    cache.delete(cache_key)
-
-    cache_key_task = f"task_{instance.task.id}"
-    cache.delete(cache_key_task)
-
-    cache.delete("all_tasks")

task_app/signals.py

@@ -0,0 +1,16 @@
+from django.core.cache import cache
+from django.db.models.signals import post_save, post_delete
+from django.dispatch import receiver
+from .models import Task, AssignedTask
+
+@receiver(post_save, sender=Task)
+@receiver(post_delete, sender=Task)
+def clear_task_cache(sender, instance, **kwargs):
+    """Invalidate cache for the affected task status and individual task"""
+    cache.delete(f"task_{instance.id}") 
+
+@receiver(post_save, sender=AssignedTask)
+@receiver(post_delete, sender=AssignedTask)
+def clear_assigned_task_cache(sender, instance, **kwargs):
+    """Invalidate cache when a task assignment changes"""
+    cache.delete(f"task_{instance.task.id}")

task_app/tests.py

@@ -2,50 +2,39 @@ from django.test import TestCase
 from django.core.cache import cache
 from django.contrib.auth import get_user_model
 from .models import Task, AssignedTask
-from .caching import get_cached_tasks, get_cached_tasks_by_status, get_cached_task_by_id
+from .caching import get_cached_task_by_id
-from .choices import TaskStatus
+from .choices import TaskStatus, TaskCategory, TaskPriority
 
 User = get_user_model()
 
 class TaskCacheTests(TestCase):
+    
     def setUp(self):
         user = User.objects.create(first_name="Test", last_name="User", email="test@example.com")
         self.user = user
+
         self.task1 = Task.objects.create(
             title="Test Task 1",
             description="Description 1",
-            category="Work",
+            category=TaskCategory.TECHNICAL_TASK,
-            priority="High",
+            priority=TaskPriority.LOW,
             status=TaskStatus.TODO,
             date="2025-01-01",
             creator=self.user
         )
+        
         self.task2 = Task.objects.create(
             title="Test Task 2",
             description="Description 2",
-            category="Personal",
+            category=TaskCategory.USER_STORY,
-            priority="Medium",
+            priority=TaskPriority.MEDIUM,
             status=TaskStatus.IN_PROGRESS,
             date="2025-01-02",
             creator=self.user
         )
+
         cache.clear()
 
-    def test_get_cached_tasks(self):
-        tasks = get_cached_tasks()
-        self.assertEqual(len(tasks), 2)
-        self.assertIn(self.task1, tasks)
-        self.assertIn(self.task2, tasks)
-
-    def test_get_cached_tasks_by_status(self):
-        tasks_todo = get_cached_tasks_by_status(TaskStatus.TODO)
-        self.assertEqual(len(tasks_todo), 1)
-        self.assertEqual(tasks_todo[0], self.task1)
-        
-        tasks_in_progress = get_cached_tasks_by_status(TaskStatus.IN_PROGRESS)
-        self.assertEqual(len(tasks_in_progress), 1)
-        self.assertEqual(tasks_in_progress[0], self.task2)
-
     def test_get_cached_task_by_id(self):
         task = get_cached_task_by_id(self.task1.id)
         self.assertIsNotNone(task)
@@ -54,62 +43,99 @@ class TaskCacheTests(TestCase):
     def test_cache_invalidation_on_task_save(self):
         cache.clear()
 
-        get_cached_tasks()
+        task = get_cached_task_by_id(self.task1.id)
+        self.assertEqual(task, self.task1)
+
         self.task1.status = TaskStatus.DONE
         self.task1.save()
 
-        self.assertIsNone(cache.get("all_tasks"))
-        self.assertIsNone(cache.get(f"tasks_by_status_{TaskStatus.TODO}"))
         self.assertIsNone(cache.get(f"task_{self.task1.id}"))
 
-        tasks = get_cached_tasks()
+        task = get_cached_task_by_id(self.task1.id)
-        self.assertIn(self.task1, tasks)
+        self.assertEqual(task, self.task1)
 
     def test_cache_invalidation_on_task_delete(self):
         cache.clear()
 
-        get_cached_tasks()
+        task = get_cached_task_by_id(self.task1.id)
+        self.assertEqual(task, self.task1)
+
         task_id = self.task1.id
         self.task1.delete()
 
-        self.assertIsNone(cache.get("all_tasks"))
         self.assertIsNone(cache.get(f"task_{task_id}"))
 
-        tasks = get_cached_tasks()
+        task = get_cached_task_by_id(task_id)
-        self.assertNotIn(self.task1, tasks)
+        self.assertIsNone(task)
 
     def test_cache_invalidation_on_assigned_task_save(self):
         cache.clear()
 
         AssignedTask.objects.create(user_id=self.user, task=self.task1)
         
-        self.assertIsNone(cache.get("all_tasks"))
-        self.assertIsNone(cache.get(f"tasks_by_status_{self.task1.status}"))
         self.assertIsNone(cache.get(f"task_{self.task1.id}"))
 
-        tasks = get_cached_tasks_by_status(self.task1.status)
+        task = get_cached_task_by_id(self.task1.id)
-        self.assertIn(self.task1, tasks)
+        self.assertEqual(task, self.task1)
 
     def test_cache_invalidation_on_assigned_task_delete(self):
         cache.clear()
 
         assigned_task = AssignedTask.objects.create(user_id=self.user, task=self.task1)
-
         assigned_task.delete()
-        
+
-        self.assertIsNone(cache.get("all_tasks"))
-        self.assertIsNone(cache.get(f"tasks_by_status_{self.task1.status}"))
         self.assertIsNone(cache.get(f"task_{self.task1.id}"))
-        
+
-        tasks = get_cached_tasks_by_status(self.task1.status)
+        task = get_cached_task_by_id(self.task1.id)
-        self.assertIn(self.task1, tasks)
+        self.assertEqual(task, self.task1)
 
     def test_cache_serialization_error_handling(self):
         cache.set("task_corrupted", b"corrupted_data")
-
         cache.delete("task_corrupted")
-
+        
         self.assertIsNone(cache.get("task_corrupted"))
 
         task = get_cached_task_by_id(self.task1.id)
-        self.assertEqual(task, self.task1)
+        self.assertEqual(task, self.task1)
+
+    def test_cache_behavior_on_multiple_tasks(self):
+        cache.clear()
+
+        task1 = get_cached_task_by_id(self.task1.id)
+        task2 = get_cached_task_by_id(self.task2.id)
+
+        self.assertEqual(task1, self.task1)
+        self.assertEqual(task2, self.task2)
+
+    def test_cache_removal_on_multiple_task_deletion(self):
+        cache.clear()
+
+        task1_id = self.task1.id
+        task2_id = self.task2.id
+
+        self.task1.delete()
+        self.task2.delete()
+
+        self.assertIsNone(cache.get(f"task_{task1_id}"))
+        self.assertIsNone(cache.get(f"task_{task2_id}"))
+
+    def test_cache_after_task_update(self):
+        cache.clear()
+
+        self.task1.title = "Updated Task 1"
+        self.task1.save()
+
+        task = get_cached_task_by_id(self.task1.id)
+        self.assertEqual(task.title, "Updated Task 1")
+
+    def test_cache_check_on_no_changes(self):
+        cache.clear()
+
+        task = get_cached_task_by_id(self.task1.id)
+        self.assertEqual(task, self.task1)
+
+        cached_task = cache.get(f"task_{self.task1.id}")
+        self.assertIsNotNone(cached_task)
+
+        cache.delete(f"task_{self.task1.id}")
+        self.assertIsNone(cache.get(f"task_{self.task1.id}"))

task_app/views.py

@@ -4,7 +4,7 @@ from rest_framework import status
 from .models import Task, SubTask, AssignedTask
 from .serializers import TaskSerializer, SubTaskSerializer, AssignedTaskSerializer
 from rest_framework.decorators import action
-from .caching import get_cached_tasks, get_cached_task_by_id, get_cached_tasks_by_status
+from .caching import get_cached_task_by_id
 from django.core.cache import cache
 from .choices import TaskStatus
 from rest_framework.permissions import IsAuthenticated
@@ -15,15 +15,6 @@ class TaskViewSet(viewsets.ModelViewSet):
     queryset = Task.objects.all()
     serializer_class = TaskSerializer
 
-    def list(self, request, *args, **kwargs):
-        status_param = request.query_params.get('status', None)
-        if status_param:
-            tasks = get_cached_tasks_by_status(status_param)
-        else:
-            tasks = get_cached_tasks()
-        serializer = self.get_serializer(tasks, many=True)
-        return Response(serializer.data)
-
     def retrieve(self, request, pk=None):
         task = get_cached_task_by_id(pk)
         if not task:
@@ -35,7 +26,6 @@ class TaskViewSet(viewsets.ModelViewSet):
         serializer = self.get_serializer(data=request.data)
         if serializer.is_valid():
             task = serializer.save()
-            cache.delete("tasks")
             cache.delete(f"task_{task.id}")
             return Response(serializer.data, status=status.HTTP_201_CREATED)
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@@ -45,7 +35,6 @@ class TaskViewSet(viewsets.ModelViewSet):
             task = Task.objects.get(pk=pk)
             task.delete()
             cache.delete(f"task_{pk}")
-            cache.delete("tasks")
             return Response({'message': 'Task deleted successfully.'}, status=status.HTTP_204_NO_CONTENT)
         except Task.DoesNotExist:
             return Response({'error': 'Task not found.'}, status=status.HTTP_404_NOT_FOUND)
@@ -63,13 +52,11 @@ class TaskViewSet(viewsets.ModelViewSet):
             return Response({'error': f'Invalid status. Valid statuses: {valid_statuses}'}, status=status.HTTP_400_BAD_REQUEST)
 
         cache.delete(f"task_{task.id}")
-        cache.delete(f"tasks_by_status_{task.status}")
-        cache.delete(f"tasks_by_status_{new_status}") 
 
         task.status = new_status
         task.save()
         return Response({'status': 'Status updated.'})
-    
+
 class SubTaskViewSet(viewsets.ModelViewSet):
     permission_classes = [IsAuthenticated]
     

user_app/caching.py

@@ -1,31 +1,34 @@
 import json
 from django.core.cache import cache
-from .models import User
 from django.core.exceptions import ObjectDoesNotExist
+from .models import User
 from .serializers import UserSerializer
 
-def get_cached_user(user_id):
+CACHE_TIMEOUT = 3600
-    cache_key = f"user_{user_id}"
+
-    cached_user = cache.get(cache_key)
+def get_cached_user_by_id(user_id):
+    return _get_or_set_cache(f"user_{user_id}", user_id)
+
+def _get_or_set_cache(cache_key, user_id=None):
+    cached_json = cache.get(cache_key)
     
-    if cached_user is not None:
+    if cached_json:
-        return cached_user
+        try:
+            return json.loads(cached_json)
+        except json.JSONDecodeError:
+            cache.delete(cache_key)
+    
+    return _fetch_and_cache(cache_key, user_id)
+
+def _fetch_and_cache(cache_key, user_id):
     try:
         user = User.objects.get(id=user_id)
     except ObjectDoesNotExist:
         return None
-
+    
     serializer = UserSerializer(user)
     user_data = serializer.data
- 
+    
-    cache.set(cache_key, user_data, timeout=3600)
+    cache.set(cache_key, json.dumps(user_data), timeout=CACHE_TIMEOUT)
-
+    
-    return user_data
+    return user_data
-
-
-def cache_user(user_id, user_data, timeout=3600):
-    """
-    Saves user data in the cache.
-    """
-    cache_key = f"user_{user_id}" 
-    cache.set(cache_key, user_data, timeout)

user_app/signals.py

@@ -7,5 +7,4 @@ from .models import User
 @receiver(post_delete, sender=User)
 def clear_user_cache(sender, instance, **kwargs):
     """Invalidate cache for the affected user"""
-    cache_key = f"user_{instance.id}"
+    cache.delete(f"user_{instance.id}") 
-    cache.delete(cache_key)

user_app/tests.py

@@ -1,16 +1 @@
 from django.test import TestCase
-from .models import User
-from .caching import get_cached_user
-
-class UserCacheTest(TestCase):
-    def test_cache_invalidierung_nach_speichern(self):
-        user = User.objects.create(first_name="Test", last_name="User", email="test@example.com")
-
-        cached_user = get_cached_user(user.id)
-        self.assertIsNotNone(cached_user)
-
-        user.first_name = "Updated"
-        user.save()
-
-        cached_user_after_update = get_cached_user(user.id)
-        self.assertIsNotNone(cached_user_after_update)

user_app/views.py

@@ -2,25 +2,24 @@ from rest_framework import viewsets
 from rest_framework.response import Response
 from .models import User
 from .serializers import UserSerializer
-from .caching import get_cached_user, cache_user
+from .caching import get_cached_user_by_id
 from rest_framework.permissions import IsAuthenticated
+from rest_framework import status
 
 class UserViewSet(viewsets.ModelViewSet):
     permission_classes = [IsAuthenticated]
     
     queryset = User.objects.all()
     serializer_class = UserSerializer
+    
+    def retrieve(self, request, pk=None):
+        user = get_cached_user_by_id(pk)
 
-    def retrieve(self, request, *args, **kwargs):
+        if not user:
-        user_id = kwargs.get("pk") 
+            try:
-        cached_user = get_cached_user(user_id)
+                user = User.objects.get(pk=pk)
+            except User.DoesNotExist:
+                return Response({'error': 'User not found.'}, status=status.HTTP_404_NOT_FOUND)
         
-        if cached_user:
+        serializer = self.get_serializer(user)
-            return Response(cached_user)
+        return Response(serializer.data)
-
-        user = self.get_object()
-        serialized_user = UserSerializer(user).data
-        
-        cache_user(user_id, serialized_user)
-
-        return Response(serialized_user)