django-business-logic

name	django-business-logic
description	services.py, selectors.py, Celery tasks, Signals, and background processing

Django Business Logic

This skill covers how to safely handle background tasks, decouple side-effects, manage retries, and avoid the implicit control flow that Signals introduce.

---

1. Celery: Architecture and Task Design

The Thin Task / Fat Service Principle

A Celery task is not a place to write business logic. It is an execution boundary: it retrieves objects from the database, delegates to a service, and handles Celery-specific concerns (retries, acknowledgment, etc.).

Why? Because:

• You need to be able to call the service directly from a management command, test, or REPL without spinning up a Celery worker.
• Business logic inside a task cannot be easily unit-tested (Celery's test helpers are cumbersome).
• Celery task functions are not strongly typed; service functions are.

✅ Recommended pattern:

# apps/reports/tasks.py
from celery import shared_task
from celery.utils.log import get_task_logger
from django.db import OperationalError

from apps.reports.models import Report
from apps.reports.services import generate_report_service

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,    # seconds
    acks_late=True,            # Only ack AFTER the task completes (prevents message loss on crash)
    reject_on_worker_lost=True,
)
def generate_report_task(self, report_id: int) -> None:
    logger.info("Starting report generation", extra={"report_id": report_id})
    
    try:
        report = Report.objects.get(id=report_id)
    except Report.DoesNotExist:
        logger.warning(f"Report {report_id} not found, skipping.")
        return  # Non-retriable: the record is gone

    try:
        generate_report_service(report=report)
    except OperationalError as exc:
        # Transient DB error — retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 10)
    except Exception as exc:
        logger.exception(f"Failed to generate report {report_id}")
        raise  # Re-raise so Celery marks task as FAILURE

Passing IDs, Not Objects

Always pass primary keys to tasks, never model instances. Model instances are serialized with pickle, which is fragile, version-sensitive, and creates large payloads.

❌ Anti-pattern:

generate_report_task.delay(report=report_instance)  # Serializes entire object

✅ Recommended:

generate_report_task.delay(report_id=report.id)    # Just the PK — re-fetch inside task

acks_late=True and Idempotency

acks_late=True means the message is only removed from the queue after the task successfully completes. If the worker crashes mid-task, the message is re-queued and the task re-runs.

This means your tasks must be idempotent — running them twice must produce the same result.

✅ Recommended idempotency pattern:

def generate_report_task(self, report_id: int) -> None:
    report = Report.objects.get(id=report_id)
    
    if report.status == Report.Status.COMPLETED:
        logger.info(f"Report {report_id} already completed, skipping.")
        return  # Safe to skip — already done

    generate_report_service(report=report)

Celery Beat (Periodic Tasks)

For scheduled jobs, use django-celery-beat to store schedules in the database rather than hardcoding them in CELERYBEAT_SCHEDULE.

# config/celery.py
app.conf.beat_scheduler = "django_celery_beat.schedulers:DatabaseScheduler"

---

2. Django Signals: When to Use and When to Refuse

Django's signal system (post_save, pre_delete, m2m_changed) implements the Observer pattern. But implicit control flow is the leading cause of "action-at-a-distance" bugs in Django applications.

The Core Problem with Signals for Business Logic

When you connect business logic to a signal, every future Model.objects.create() or .save() anywhere in the codebase silently triggers that logic — including in migrations, tests, management commands, and background tasks where you may not want it.

❌ Anti-pattern (Business Logic in Signal):

# apps/users/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
    if created:
        Profile.objects.create(user=instance)
        send_welcome_email(instance.email)   # Side effect hidden from callers!
        analytics.track("user_signed_up", user_id=instance.id)

The problem:

1. A developer calls User.objects.create(email="[e-mail]") in a test and accidentally triggers an email.
2. A data migration creates users and sends thousands of welcome emails to production users.
3. The Profile.objects.create() can fail and raise, rolling back the parent transaction in unexpected places.
4. Debugging requires tracing through signals.py, receivers.py, and apps.py to understand what happens on .save().

✅ Recommended (Explicit Service):

# apps/users/services.py
@transaction.atomic
def register_user(*, email: str, password: str) -> User:
    user = User.objects.create_user(email=email, password=password)
    Profile.objects.create(user=user)
    transaction.on_commit(lambda: send_welcome_email_task.delay(user_id=user.id))
    transaction.on_commit(lambda: track_signup_task.delay(user_id=user.id))
    return user

Now the full action is readable in one function. Tests can mock send_welcome_email_task. Migrations can call User.objects.create() without triggering any side effects.

Acceptable Use Cases for Signals

Use signals only for cross-cutting concerns that are genuinely orthogonal to domain logic — i.e., things that should happen for every instance save regardless of context:

Use Case	Signal	Reason
Invalidate cache on save	post_save, post_delete	Infrastructure concern, not business logic
Update Elasticsearch index	post_save	Cross-cutting, no domain side effects
Write to audit log	post_save	Observational, no side effects that affect the domain
Bust CDN cache	post_save	Infrastructure

Even for these, consider using django-lifecycle (which hooks directly on model methods) instead of signal receivers, for better traceability.

---

3. Handling Long-Running Workflows

For multi-step processes (e.g., onboarding, order processing), resist the temptation to chain Celery tasks with .chain() or .chord(). These are difficult to debug, monitor, and retry at specific steps.

✅ Recommended: Explicit State Machine on the Model

class Order(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending"
        PAYMENT_PROCESSING = "payment_processing"
        INVENTORY_RESERVED = "inventory_reserved"
        SHIPPED = "shipped"
        FAILED = "failed"

    status = models.CharField(max_length=30, choices=Status.choices, default=Status.PENDING)

Each Celery task transitions to the next status and enqueues the next task:

@shared_task(bind=True, acks_late=True)
def process_order_payment_task(self, order_id: int):
    order = Order.objects.select_for_update().get(id=order_id)
    
    if order.status != Order.Status.PENDING:
        return  # Idempotency guard
    
    process_payment_service(order=order)  # Sets status to PAYMENT_PROCESSING
    
    transaction.on_commit(
        lambda: reserve_inventory_task.delay(order_id=order.id)
    )

This gives you full visibility: query Order.objects.filter(status="payment_processing") to see stuck orders. Each step is individually retryable.

Navigation

• Core Architecture
• Models & ORM
• Views & APIs