django-tenants-temporal

Tenant-aware Temporal workflows and activities for django-tenants.

If you have used tenant-schemas-celery, this is the Temporal equivalent — and there wasn't one, which is why this exists.

The mental model. The schema active when you start a workflow is written into a Temporal header. Every hop after that carries the header along and binds it to a contextvar. Before a sync activity's body runs, the schema is re-entered on that worker thread's connection. Your activities keep using the ORM exactly as they would inside a request, and no workflow signature ever mentions a tenant.

┌─ Django process ──────────────────────────────────┐
│  schema "acme" is active                          │
│  start_workflow()  ──►  header: schema = "acme"   │
└──────────────────────────┬────────────────────────┘
                           │  Temporal server carries the header
                           ▼
┌─ TenantWorker ────────────────────────────────────┐
│  TenantSchemaInterceptor installed                │
│                                                   │
│   Workflow  (sandboxed — no Django, no ORM)       │
│     header ──► contextvar                         │
│        │                                          │
│        │  child workflow · signal · query ·       │
│        │  update · continue-as-new                │
│        │  each one re-stamped with "acme"         │
│        ▼                                          │
│   Sync activity  (worker thread)                  │
│     schema re-entered on this thread's connection │
│     your ORM calls — tenant already active        │
└───────────────────────────────────────────────────┘

Install

pip install django-tenants-temporal

Requires django-tenants>=3.5, django>=4.2, temporalio>=1.9.

Quick start

1. Client — add the interceptor wherever you build a Client:

from django_tenants_temporal import connect

client = await connect("localhost:7233", namespace="default")

Or on a Client you build yourself:

from temporalio.client import Client
from django_tenants_temporal import TenantSchemaInterceptor

client = await Client.connect("localhost:7233", interceptors=[TenantSchemaInterceptor()])

2. Worker — TenantWorker wraps your activities and installs the interceptor:

from concurrent.futures import ThreadPoolExecutor
from django_tenants_temporal import TenantWorker, autodiscover

workflows, activities = autodiscover()
worker = TenantWorker(
    client,
    task_queue="default",
    workflows=workflows,
    activities=activities,
    activity_executor=ThreadPoolExecutor(max_workers=8),
)

Or skip the wiring entirely:

python manage.py run_temporal_worker --task-queue default

autodiscover() walks INSTALLED_APPS for <app>/workflows/ and <app>/activities/ packages, the way Celery's autodiscover_tasks walks for tasks.py.

3. Dispatch — from ordinary synchronous Django code:

from django_tenants_temporal import start_workflow

def my_view(request):
    start_workflow(SendRemindersWorkflow, invoice.id, task_queue="default")

The schema is captured from connection.schema_name on the request thread. There is no step 4 — activities need no import from this package.

How this differs from tenant-schemas-celery

tenant-schemas-celery smuggles the schema through the task's arguments, because that is what Celery offers. Temporal has first-class headers, which buys three things:

• Call signatures stay clean. Nothing is injected into your arguments.
• It survives the hops Celery doesn't have. Child workflows, signals, queries, updates and continue_as_new all keep the tenant.
• Workflows stay deterministic. The schema is carried through workflow code, never used by it. Only activities touch the database.

Settings

All optional; the defaults are what most projects want.

DJANGO_TENANTS_TEMPORAL = {
    "header_key": "__tenant_schema",
    "default_schema": None,
    "validate_tenant": True,
    "tenant_cache_seconds": 0,
    "close_old_connections": True,
    "strict": True,
}

Key	Default	Meaning
header_key	"__tenant_schema"	Name of the Temporal header carrying the schema.
default_schema	None	Schema to use when no header arrived. None leaves the connection untouched.
validate_tenant	True	Check the schema exists and has a tenant row before entering it.
tenant_cache_seconds	0	Cache the tenant lookup per schema. 0 disables caching.
close_old_connections	True	Recycle stale connections around each activity. See below.
strict	True	A missing tenant raises a non-retryable error. False runs the activity unscoped.

An unknown key raises ConfigError at startup with a spelling suggestion, rather than silently doing nothing.

Why close_old_connections matters

Django recycles database connections at request boundaries. Celery gets the same treatment from Django's signal hooks. Temporal has neither — worker threads are long-lived, so a CONN_MAX_AGE connection the database has since dropped will resurface as InterfaceError on a worker that has been idle. This package brackets every activity with close_old_connections() so that cannot happen. Leave it on unless you have a specific reason.

Workflow sandbox

Importing anything from this package inside a workflow file trips Temporal's sandbox:

temporalio.worker.workflow_sandbox._restrictions.RestrictedWorkflowAccessError:
Cannot access django.db.connection from inside a workflow.

Pass the package through:

from django_tenants_temporal import tenant_sandbox_runner

worker = TenantWorker(..., workflow_runner=tenant_sandbox_runner())

tenant_sandbox_runner("myapp.shared") passes extra modules through too. The package is safe to pass through: the modules the sandbox loads (context, interceptor) import no Django and hold no mutable state beyond a contextvar — there is a test asserting exactly that.

The better habit is to keep workflow files free of Django imports and put ORM access in activities, importing Django inside the function body. See tests/testapp/temporal_defs.py.

Schedules

Client.create_schedule is not on the interceptor chain, so scheduled workflows need the header attached explicitly:

from temporalio.client import ScheduleActionStartWorkflow
from django_tenants_temporal import tenant_schedule_action

action = tenant_schedule_action(
    ScheduleActionStartWorkflow(NightlyWorkflow.run, id="nightly", task_queue="default"),
    schema="acme",
)
await client.create_schedule("nightly-acme", Schedule(action=action, spec=ScheduleSpec(...)))

This is the analogue of Celery beat's tenant-aware schedulers: one schedule per tenant, each stamped with its own schema.

Testing

Eager mode runs workflow bodies inline, calling activities directly — no server, no worker:

from django_tenants_temporal.testing import eager_mode, run_workflow_eagerly

def test_reminders(db):
    assert run_workflow_eagerly(SendRemindersWorkflow, invoice.id, schema="acme") == 1

It lives in django_tenants_temporal.testing, not the package root, because it monkey-patches temporalio.workflow.execute_activity. Importing it is a statement that you are in a test.

Eager mode is deliberately faithful to production: activities are wrapped exactly as TenantWorker wraps them, so the schema is entered, and they run in a worker thread rather than on the event loop, so sync ORM calls behave the same way. Without both, an eager test could pass while the real worker wrote to the wrong tenant.

For real end-to-end coverage use temporalio.testing.WorkflowEnvironment.start_local(); see tests/test_integration.py.

Limitations / help wanted

Two things are deliberately out of scope for v0.1. Both are good contributions and both have open issues.

Async activities are not tenant-aware

@activity.defn async def activities run fine and can read current_schema(), but nothing enters the schema for them — the wrapper passes them through and logs a warning at registration. Sync activities are the supported path.

Why it isn't a one-liner: schema_context issues a per-connection SET search_path, and Django's connections are thread-local. An async activity runs on the event loop, so every ORM call would have to be funnelled through sync_to_async(thread_sensitive=False) onto a thread whose connection was actually switched. That deserves a deliberate design rather than a wrapper that appears to work.

Workaround today:

@activity.defn
async def my_activity() -> int:
    schema = current_schema()

    def work():
        with schema_context(schema):
            return Invoice.objects.count()

    return await sync_to_async(work, thread_sensitive=False)()

Single database only

There is no databases setting yet — the analogue of tenant-schemas-celery's tenant_databases. schema_context targets get_tenant_database_alias() and nothing else.

A fix would wrap the activation in an ExitStack over the configured aliases, using connection.set_schema() / restore for the non-default ones. Adding it is purely additive, so it will not break the current API.

Caveats

• Workflows must not touch the ORM. They carry the schema; they don't use it. Database work belongs in activities. This is a Temporal determinism rule, not our restriction.
• Deleted tenants. With validate_tenant on, an activity for a schema that no longer exists fails with a non-retryable ApplicationError rather than retrying against a missing schema.
• Shared worker vs worker-per-tenant. A single worker serves every tenant; the schema is per-call, not per-worker. Run a worker per tenant only if you need resource isolation.
• CONN_MAX_AGE. Works as expected, but see the note on close_old_connections above.

Contributing

See CONTRIBUTING.md. docker compose up -d postgres, then pytest -m "not integration" for the fast suite.

License

MIT.