ICVOSS DJANGO PACKAGE REGISTRY

The package index django-boundary Run Celery tasks with tenant context

Run Celery tasks with tenant context

Documentation

Goal

Make Celery tasks run under the same tenant that dispatched them, so that tenant-scoped queries inside the task filter correctly without you passing a tenant ID through task arguments.

Prerequisites

How it works

boundary propagates the tenant through Celery task headers, not task arguments:

  1. At dispatch, the active tenant's primary key is serialised into two headers: boundary_tenant_id (exposed as HEADER_TENANT_ID) and, when the tenant has a region, boundary_region (exposed as HEADER_REGION).
  2. On the worker, before the task body runs, the headers are read and the tenant is loaded from the database and pushed into TenantContext. The context is cleared again once the task finishes.

Because the tenant travels in headers, your task signatures stay clean and you never risk leaking the tenant ID into task arguments or result backends in an unexpected shape.

There are two ways to wire this in: the TenantTask base class (handles both dispatch and worker sides) and the @tenant_task decorator (handles the worker side only).

Steps

TenantTask handles both sides: it injects headers in apply_async and restores context in __call__. Mix it with your Celery app's Task class.

  1. Define the task using TenantTask as the first base:

```python # bookings/tasks.py from celery import shared_task

from boundary.celery import TenantTask from myproject.celery import app # your Celery() instance

class GenerateReport(TenantTask, app.Task): def run(self, report_id): # Tenant context is already active here. # Tenant-scoped queries filter automatically. report = Report.objects.get(id=report_id) report.build()

GenerateReport = app.register_task(GenerateReport()) ```

  1. Dispatch it from anywhere that has an active tenant:

```python from boundary.context import TenantContext

with TenantContext.using(club): GenerateReport.delay(report_id=42) # headers boundary_tenant_id (and boundary_region, if set) # are injected automatically at dispatch time ```

In a request handled by TenantMiddleware, the tenant is already in context, so a bare GenerateReport.delay(report_id=42) picks it up.

Option B: @tenant_task decorator (worker side only)

Use the decorator when you cannot change the task base class, for example when you want a plain function task. The decorator restores context on the worker but does not inject headers at dispatch. You must arrange header injection yourself (typically by also using TenantTask for apply_async, or via your own Celery before_task_publish signal that calls into the header helpers).

  1. Decorate the task function, placing @tenant_task below @app.task:

```python # bookings/tasks.py from boundary.celery import tenant_task from myproject.celery import app

@app.task @tenant_task def send_confirmation(booking_id): # Tenant context is restored from headers before this runs. booking = Booking.objects.get(id=booking_id) booking.send_confirmation_email() ```

Verify it worked

Inspect the headers produced for the active tenant. This is the exact logic that runs at dispatch:

from boundary.celery import HEADER_TENANT_ID, HEADER_REGION, _get_tenant_headers
from boundary.context import TenantContext

with TenantContext.using(club):
    headers = _get_tenant_headers()

assert headers[HEADER_TENANT_ID] == str(club.pk)
# HEADER_REGION is present only when the tenant has a region value

Inside the running task, confirm the tenant is active:

from boundary.context import TenantContext

def run(self, report_id):
    assert TenantContext.get() is not None
    ...

For tests, use the set_tenant helper from boundary.testing to establish a tenant before dispatching:

from boundary.testing import set_tenant

with set_tenant(tenant_a):
    headers = _get_tenant_headers()
assert headers[HEADER_TENANT_ID] == str(tenant_a.pk)

Common pitfalls