Tasks Signals - Architecture Explanation

File: api/nextango/apps/tasks/signals.py Status: Does not exist Purpose: Explain why Tasks domain has no signals.py

Overview

The Tasks domain intentionally has no signals.py file. This is an architectural decision based on the domain's role as a lightweight operational domain with minimal cross-domain integration needs.

Key Principle:

Business Domain (users, products, stores, transactions, promotions)
    │
    ├─ Signals for Sanity CMS sync
    ├─ Complex cross-domain interactions
    ├─ State change notifications
    │
    ▼
Infrastructure Domain (tasks, web_auth, integrations)
    │
    ├─ Minimal signals (or none)
    ├─ Self-contained logic
    ├─ Limited cross-domain dependencies

Why Tasks Has No Signals

1. Minimal Cross-Domain Integration

Tasks Domain Characteristics:

  • Internal operational tool (not customer-facing)
  • Store-scoped (doesn't affect inventory, transactions, or customers)
  • Optional Sanity sync (not critical for business operations)

Comparison with Business Domains:

Products Domain (Has Signals):

# products/signals.py
@receiver(post_save, sender=Product)
def sync_product_to_sanity(sender, instance, created, **kwargs):
    """
    Critical: Product changes must sync to CMS
    - Updates POS inventory displays
    - Affects customer-facing catalog
    - Triggers price updates across systems
    """
    product_sync_handler.sync_to_sanity(instance)

Tasks Domain (No Signals Needed):

# No signals.py
# Why? Task creation/updates don't affect:
# - Inventory levels
# - Transaction processing
# - Customer experience
# - Cross-domain state

2. Optional Sanity CMS Integration

Pattern: Tasks CAN sync to Sanity but don't NEED to

If Sanity Sync Required:

# tasks/signals.py (optional implementation)
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Task

@receiver(post_save, sender=Task)
def sync_task_to_sanity(sender, instance, created, **kwargs):
    """Optional: Sync tasks to Sanity CMS for centralized management."""
    from nextango.apps.integrations.utils.sync_context import should_skip_sync_to_sanity

    if should_skip_sync_to_sanity():
        return  # Prevent infinite loops from webhook processing

    try:
        from nextango.apps.integrations.sync_services import task_sync_handler
        task_sync_handler.sync_to_sanity(instance)
    except Exception as e:
        logger.error(f"Failed to sync task {instance.task_id} to Sanity: {e}")

Current State: Not implemented because:

  • Tasks are primarily created/managed in Django (not Sanity)
  • Task management is internal operations tool
  • Sanity integration not required for task functionality

3. Service Layer Handles Business Logic

Pattern: Business logic in services, not signals

TaskService handles:

# tasks/services.py
class TaskService:
    @staticmethod
    def get_task_with_details(task_id):
        """Business logic for task retrieval with computed fields."""
        task = Task.objects.select_related(...).get(task_id=task_id)

        return {
            'is_overdue': task.due_date and task.due_date < date.today(),
            'days_since_created': (date.today() - task.created_at.date()).days,
            # ... other computed fields
        }

No Need for Signals:

  • Computed fields calculated on-demand (not cached)
  • No state changes to propagate
  • No external systems to notify

Comparison: Business vs Operational Domains

Business Domains (Have Signals)

Users Domain:

# users/signals.py
@receiver(post_save, sender=User)
def sync_user_to_sanity(sender, instance, created, **kwargs):
    """Critical: User changes affect authentication, transactions, permissions."""
    pass

@receiver(post_save, sender=Role)
def update_user_permissions(sender, instance, created, **kwargs):
    """Update permissions cache when roles change."""
    pass

Transactions Domain:

# transactions/signals.py
@receiver(post_save, sender=SaleTransaction)
def commit_inventory_on_completion(sender, instance, created, **kwargs):
    """Critical: Completed transactions must commit inventory."""
    if instance.status == 'completed':
        InventoryAdjustmentService.commit_reserved_inventory(instance)

Products Domain:

# products/signals.py
@receiver(post_save, sender=Product)
def sync_product_to_sanity(sender, instance, created, **kwargs):
    """Critical: Product changes sync to CMS for customer-facing catalog."""
    product_sync_handler.sync_to_sanity(instance)

Operational Domains (Minimal/No Signals)

Tasks Domain:

  • Signals: None
  • Why: Internal operations, optional Sanity sync, no critical cross-domain logic

Web Auth Domain:

  • Signals: None (likely)
  • Why: Session management is self-contained, no cross-domain state changes

Integrations Domain:

  • Signals: None (by design)
  • Why: Infrastructure layer that receives signals, doesn't emit them

When Would Tasks Need Signals?

Scenario 1: Required Sanity Sync

Trigger: Task management centralized in Sanity CMS

Implementation:

# tasks/signals.py
@receiver(post_save, sender=Task)
def sync_task_to_sanity(sender, instance, created, **kwargs):
    from nextango.apps.integrations.utils.sync_context import should_skip_sync_to_sanity

    if should_skip_sync_to_sanity():
        return

    task_sync_handler.sync_to_sanity(instance)

Scenario 2: Task Completion Triggers Automation

Trigger: Completing tasks triggers other business logic

Implementation:

# tasks/signals.py
@receiver(post_save, sender=Task)
def handle_task_completion(sender, instance, created, **kwargs):
    """Trigger automation when tasks are completed."""
    if instance.status == 'completed' and instance.priority == 'high':
        # Example: Send notification to store manager
        NotificationService.send_task_completed_notification(instance)

        # Example: Create follow-up task
        if instance.title.startswith('Daily'):
            TaskService.create_followup_task(instance)

Scenario 3: Task Analytics

Trigger: Track task metrics in real-time

Implementation:

# tasks/signals.py
@receiver(post_save, sender=Task)
def track_task_metrics(sender, instance, created, **kwargs):
    """Track task creation/completion metrics."""
    from nextango.apps.analytics.services import AnalyticsService

    if created:
        AnalyticsService.track_task_created(instance)
    elif instance.status == 'completed':
        completion_time = (instance.completed_at - instance.created_at).total_seconds()
        AnalyticsService.track_task_completed(instance, completion_time)

Alternative Patterns Used Instead

Pattern 1: Service Layer Methods

Instead of signals, use explicit service methods:

# tasks/services.py
class TaskService:
    @staticmethod
    def complete_task(task_id, completed_by_user):
        """Complete task with explicit business logic."""
        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'])

        # Explicit logic (no signals needed)
        if task.priority == 'high':
            NotificationService.notify_completion(task)

        return task

Benefits:

  • Explicit control flow (easier to debug)
  • No hidden side effects
  • Clear where logic executes

Pattern 2: View-Level Actions

Instead of signals, handle in views:

# tasks/views.py
class TaskViewSet(viewsets.ModelViewSet):
    def partial_update(self, request, *args, **kwargs):
        """Override PATCH to handle task completion logic."""
        response = super().partial_update(request, *args, **kwargs)

        task = self.get_object()
        if task.status == 'completed':
            # Explicit logic triggered by API call
            self.handle_task_completion(task)

        return response

    def handle_task_completion(self, task):
        """Business logic for task completion."""
        NotificationService.notify_completion(task)

Benefits:

  • Logic tied to API action (clearer causation)
  • Easier to test (no signal mocking needed)
  • Explicit in request/response cycle

Users Domain Pattern

# users/signals.py
@receiver(post_save, sender=User)
def sync_user_to_sanity(sender, instance, created, **kwargs):
    """Sync user changes to Sanity CMS."""
    pass

@receiver(post_save, sender=Role)
def sync_role_to_sanity(sender, instance, created, **kwargs):
    """Sync role changes to Sanity CMS."""
    pass

Why Users Needs Signals:

  • User changes affect authentication across systems
  • Role updates must propagate to permissions
  • Employee data syncs to CMS for centralized management

Transactions Domain Pattern

# transactions/signals.py
@receiver(post_save, sender=SaleTransaction)
def handle_transaction_state_changes(sender, instance, created, **kwargs):
    """Handle transaction completion and inventory commits."""
    if instance.status == 'completed':
        InventoryAdjustmentService.commit_reserved_inventory(instance)

Why Transactions Needs Signals:

  • Transaction completion triggers inventory commits
  • State changes affect multiple domains (inventory, payments, analytics)
  • Critical business logic must execute reliably

Implementation Recommendation

Current State: No Signals (Correct)

Reasoning:

  • Tasks are operational/internal
  • No critical cross-domain dependencies
  • Sanity sync optional
  • Business logic handled in services

If Signals Become Needed

Recommended Implementation:

# tasks/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Task
import logging

logger = logging.getLogger(__name__)

@receiver(post_save, sender=Task)
def sync_task_to_sanity(sender, instance, created, **kwargs):
    """
    Optional: Sync tasks to Sanity CMS.
    Only enable if task management is centralized in Sanity.
    """
    from nextango.apps.integrations.utils.sync_context import should_skip_sync_to_sanity

    # Prevent infinite loops from webhook processing
    if should_skip_sync_to_sanity():
        return

    try:
        from nextango.apps.integrations.sync_services import task_sync_handler
        task_sync_handler.sync_to_sanity(instance)
    except Exception as e:
        # Non-critical: Log error but don't fail task save
        logger.error(f"Failed to sync task {instance.task_id} to Sanity: {e}")

Registration:

# tasks/apps.py
class TasksConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'nextango.apps.tasks'

    def ready(self):
        import nextango.apps.tasks.signals  # Register signals

Was this page helpful?