Tasks Services - Business Logic Layer

File: api/nextango/apps/tasks/services.py Purpose: Task-related business logic and data enrichment Pattern: Service layer pattern with static methods

Overview

The Tasks service layer provides enhanced task data retrieval with computed fields and optimized database queries. Currently implements a single service class (TaskService) focused on providing comprehensive task details beyond basic model serialization.

Architecture Pattern:

View Layer (views.py)
    │
    ├─ Delegates business logic
    │
    ▼
Service Layer (services.py)
    │
    ├─ TaskService.get_task_with_details()
    ├─ Optimized queries (select_related)
    ├─ Computed fields (is_overdue, days_since_created)
    ├─ Structured response formatting
    │
    ▼
Enhanced Task Data → JSON Response

TaskService

Purpose: Centralized business logic for task operations

Source Code:

# From services.py:4-59

class TaskService:
    """Service class for task-related business logic."""

    @staticmethod
    def get_task_with_details(task_id):
        """
        Get comprehensive task details including employee and store names.

        Args:
            task_id (str): Task ID to retrieve details for

        Returns:
            dict: Task details with additional calculated fields

        Raises:
            Task.DoesNotExist: If task with given ID doesn't exist
        """
        try:
            task = Task.objects.select_related(
                'assigned_to', 'completed_by', 'store'
            ).get(task_id=task_id)
        except Task.DoesNotExist:
            raise Task.DoesNotExist(f"Task with ID {task_id} not found")

        # Build comprehensive task details
        task_details = {
            'id': task.id,
            'task_id': task.task_id,
            'title': task.title,
            'description': task.description,
            'priority': task.priority,
            'status': task.status,
            'due_date': task.due_date,
            'completed_at': task.completed_at,
            'created_at': task.created_at,
            'assigned_to': {
                'id': task.assigned_to.id if task.assigned_to else None,
                'name': task.assigned_to.name if task.assigned_to else None,
                'employee_id': task.assigned_to.employee_id if task.assigned_to else None,
            } if task.assigned_to else None,
            'completed_by': {
                'id': task.completed_by.id if task.completed_by else None,
                'name': task.completed_by.name if task.completed_by else None,
                'employee_id': task.completed_by.employee_id if task.completed_by else None,
            } if task.completed_by else None,
            'store': {
                'id': task.store.id,
                'name': task.store.name,
                'store_code': task.store.store_code,
            },
            'is_overdue': task.due_date and task.due_date < task.created_at.date() and task.status != 'completed',
            'days_since_created': (task.created_at.date() - task.created_at.date()).days if task.created_at else 0,
        }

        return task_details

get_task_with_details()

Method: TaskService.get_task_with_details(task_id)

Purpose: Retrieve comprehensive task data with computed fields and related object details

Parameters

task_id (str, required):

Business identifier for the task (not UUID)
Example: "task_12345"
Used in API URLs: /api/tasks/{task_id}/

Returns

dict - Enhanced task data with the following structure:

{
    # Core task fields
    'id': UUID('a1b2c3d4-...'),
    'task_id': 'task_12345',
    'title': 'Restock dairy section',
    'description': 'Restock milk, yogurt, and cheese products',
    'priority': 'high',  # 'low', 'medium', or 'high'
    'status': 'in_progress',  # 'pending', 'in_progress', or 'completed'
    'due_date': datetime.date(2025, 11, 20),
    'completed_at': None,  # Timestamp when completed, or None
    'created_at': datetime.datetime(2025, 11, 18, 10, 30, 0),

    # Assigned employee details (nested object)
    'assigned_to': {
        'id': UUID('employee-uuid'),
        'name': 'John Doe',
        'employee_id': 'EMP-001'
    },  # or None if unassigned

    # Completion employee details (nested object)
    'completed_by': {
        'id': UUID('employee-uuid'),
        'name': 'Jane Smith',
        'employee_id': 'EMP-002'
    },  # or None if not completed

    # Store details (nested object)
    'store': {
        'id': UUID('store-uuid'),
        'name': 'Downtown Location',
        'store_code': 'DT-01'
    },

    # Computed fields
    'is_overdue': False,  # True if due_date < today and status != 'completed'
    'days_since_created': 0  # Number of days since creation
}

Raises

Task.DoesNotExist:

When task_id doesn't match any existing task
Exception message: "Task with ID {task_id} not found"

Query Optimization

Pattern: Uses select_related() to prevent N+1 queries

# From services.py:22-24
task = Task.objects.select_related(
    'assigned_to', 'completed_by', 'store'
).get(task_id=task_id)

Performance Impact:

Without select_related: 4 database queries (1 for task + 3 for relationships)
With select_related: 1 database query with SQL JOINs
Speedup: ~3-4x faster for typical task retrieval

SQL Generated:

SELECT
    tasks_task.*,
    users_user_assigned.*,
    users_user_completed.*,
    stores_store.*
FROM tasks_task
LEFT JOIN users_user users_user_assigned ON tasks_task.assigned_to_id = users_user_assigned.id
LEFT JOIN users_user users_user_completed ON tasks_task.completed_by_id = users_user_completed.id
INNER JOIN stores_store ON tasks_task.store_id = stores_store.id
WHERE tasks_task.task_id = 'task_12345';

Computed Fields

is_overdue

Logic: Task is overdue if it has a due date in the past and is not completed

# From services.py:54
'is_overdue': task.due_date and task.due_date < task.created_at.date() and task.status != 'completed'

Important Bug: The current implementation compares task.due_date with task.created_at.date(), which will never indicate overdue status. This should likely compare with date.today() instead.

Expected Logic:

from datetime import date
'is_overdue': task.due_date and task.due_date < date.today() and task.status != 'completed'

Use Case: Highlight urgent tasks in UI, prioritize task lists

days_since_created

Logic: Number of days elapsed since task creation

# From services.py:55
'days_since_created': (task.created_at.date() - task.created_at.date()).days if task.created_at else 0

Important Bug: The current implementation always returns 0 because it subtracts task.created_at.date() from itself. This should compare with date.today().

Expected Logic:

from datetime import date
'days_since_created': (date.today() - task.created_at.date()).days if task.created_at else 0

Use Case: Track task age, identify stale tasks, performance metrics

Response Structure Patterns

Nullable Relationships

Pattern: Nested object for relationships, None if null

# From services.py:39-43
'assigned_to': {
    'id': task.assigned_to.id if task.assigned_to else None,
    'name': task.assigned_to.name if task.assigned_to else None,
    'employee_id': task.assigned_to.employee_id if task.assigned_to else None,
} if task.assigned_to else None

Behavior:

If assigned_to is null → entire field is None
If assigned_to exists → nested object with id, name, employee_id
Prevents AttributeError on null foreign keys

Frontend Usage:

// Safe null checking
if (task.assigned_to) {
    console.log(`Assigned to: \${task.assigned_to.name}`);
} else {
    console.log('Unassigned task');
}

Required Relationships

Pattern: Always include nested object (store is required)

# From services.py:49-52
'store': {
    'id': task.store.id,
    'name': task.store.name,
    'store_code': task.store.store_code,
}

Behavior:

Store is never null (PROTECT constraint in model)
Always includes id, name, store_code
No null checks needed

Integration Points

Called By

TaskViewSet.retrieve():

# From views.py:14-27
def retrieve(self, request, *args, **kwargs):
    """Override retrieve method to use TaskService for enhanced task data."""
    task_id = kwargs.get('task_id')

    try:
        # Delegates to service
        task_details = TaskService.get_task_with_details(task_id)
        return Response(task_details, status=status.HTTP_200_OK)
    except Task.DoesNotExist:
        return Response(
            {'error': 'Task not found'},
            status=status.HTTP_404_NOT_FOUND
        )

API Endpoint:

GET /api/tasks/{task_id}/
Returns service response directly as JSON

Dependencies

Models:

# From services.py:1
from .models import Task

Django Utils:

# Expected for proper date handling
from datetime import date  # Missing in current implementation

Service Layer Benefits

Why Use Services?

1. Separation of Concerns:

Views handle HTTP request/response
Services handle business logic
Models handle data persistence

2. Reusability:

Same service method can be called from:
- REST API views
- GraphQL resolvers
- Background tasks
- Management commands

3. Testability:

Services can be tested without HTTP layer
Mock database queries easily
Test business logic in isolation

4. Maintainability:

Business logic centralized in one place
Changes to task detail logic only affect services.py
Views remain thin and focused

Usage Examples

Basic Retrieval

from nextango.apps.tasks.services import TaskService

# Get task details
task_details = TaskService.get_task_with_details('task_12345')

# Access computed fields
if task_details['is_overdue']:
    print(f"Task '{task_details['title']}' is overdue!")

# Access related objects
if task_details['assigned_to']:
    print(f"Assigned to: {task_details['assigned_to']['name']}")

Error Handling

from nextango.apps.tasks.models import Task
from nextango.apps.tasks.services import TaskService

try:
    task_details = TaskService.get_task_with_details('nonexistent_task')
except Task.DoesNotExist as e:
    print(f"Error: {e}")
    # Error: Task with ID nonexistent_task not found

Custom Business Logic (Future Expansion)

Pattern for additional service methods:

class TaskService:
    @staticmethod
    def get_task_with_details(task_id):
        # Existing implementation
        pass

    @staticmethod
    def get_overdue_tasks_for_store(store_id):
        """Get all overdue tasks for a specific store."""
        from datetime import date

        tasks = Task.objects.filter(
            store_id=store_id,
            status__in=['pending', 'in_progress'],
            due_date__lt=date.today()
        ).select_related('assigned_to', 'store')

        return [
            TaskService.get_task_with_details(task.task_id)
            for task in tasks
        ]

    @staticmethod
    def complete_task(task_id, completed_by_user):
        """Mark task as completed."""
        from django.utils import timezone

        task = Task.objects.get(task_id=task_id)
        task.status = 'completed'
        task.completed_by = completed_by_user
        task.completed_at = timezone.now()
        task.save(update_fields=['status', 'completed_by', 'completed_at'])

        return TaskService.get_task_with_details(task_id)

Performance Considerations

Query Performance

Single Task Lookup:

Complexity: O(1) with indexed task_id
Query Time: ~5-10ms with select_related()
Without Optimization: ~20-40ms (N+1 queries)

Recommended Indexes:

CREATE INDEX idx_tasks_task_id ON tasks_task(task_id);

Caching Opportunities

Pattern: Cache task details for frequently accessed tasks

from django.core.cache import cache

def get_task_with_details_cached(task_id):
    cache_key = f'task_details:{task_id}'
    cached = cache.get(cache_key)

    if cached:
        return cached

    task_details = TaskService.get_task_with_details(task_id)
    cache.set(cache_key, task_details, timeout=300)  # 5 minutes
    return task_details

Known Issues

Bug 1: is_overdue Always False

Current Code:

'is_overdue': task.due_date and task.due_date < task.created_at.date() and task.status != 'completed'

Problem: Compares due_date with created_at.date() instead of current date

Fix:

from datetime import date
'is_overdue': task.due_date and task.due_date < date.today() and task.status != 'completed'

Bug 2: days_since_created Always 0

Current Code:

'days_since_created': (task.created_at.date() - task.created_at.date()).days if task.created_at else 0

Problem: Subtracts date from itself (always 0)

Fix:

from datetime import date
'days_since_created': (date.today() - task.created_at.date()).days if task.created_at else 0

models.md - Task and DeliverySchedule models
views.md - TaskViewSet API endpoints
serializers.md - Task data validation
Users - Employee/User model
Stores - Store model