Django Logic is a workflow library for Django. You declare the states of a model, the transitions between them, and the code each transition runs. The declaration lives in one place, away from views, models and forms.

Work that is slow, external or retriable is a background transition. django-logic saves it as a database row. A worker process claims the row and runs it.

Requirements

• Python 3.11 or later.
• Django 4.2 or later. CI tests 4.2, 5.1, 5.2 and 6.0. Django 5.0 is not supported.
• PostgreSQL for background transitions. The worker claims a row with SELECT FOR UPDATE SKIP LOCKED, and SQLite has no row locks.
• A cross-process default cache. The web processes and the worker processes share the state lock through it. django-logic locks through Django's cache API and imports no backend, so django.core.cache.backends.redis.RedisCache is enough. manage.py check refuses a per-process cache (locmem, dummy) when DEBUG=False and a background transition is bound.

Install

pip install django-logic

Add one entry to INSTALLED_APPS and create the table:

INSTALLED_APPS = [
    ...,
    'django_logic',
]

That one entry is the whole library: the TransitionMessage table and its migrations, the dl_worker and dl_transitions commands, and every system check. There is nothing else to install or configure.

Running no background transitions? The same entry. The PostgreSQL and shared-cache rules below apply only when a background transition is bound — manage.py check tells you the moment one is — so a synchronous-only project runs on SQLite with Django's default cache.

Upgrading from 1.0.x or earlier? Replace 'django_logic.background' (and a second 'django_logic' line, if present) with the one entry above. Nothing in the database moves: the app keeps the same label, table and migration history, so an install that had the background app applies no new migration. A 1.0.x install of 'django_logic' alone had no table, so its next migrate creates it, empty.

python manage.py migrate

Point the default cache at a backend the web processes and the workers share:

CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': os.environ['REDIS_URL'],
    }
}

Every DJANGO_LOGIC key has a default, so you need no other configuration to start.

Declare a process

A process lists the transitions of one state field. A transition names the states it starts from, the state it ends in, and the functions it runs.

# models.py
from django.db import models


class Order(models.Model):
    STATUS_CHOICES = [
        ('draft', 'Draft'),
        ('approved', 'Approved'),
        ('fulfilled', 'Fulfilled'),
        ('fulfilment_failed', 'Fulfilment failed'),
        ('cancelled', 'Cancelled'),
    ]
    status = models.CharField(max_length=32, choices=STATUS_CHOICES, default='draft')

# process.py
from django_logic import Process, Transition


def has_stock(instance, **kwargs):
    return all(item.product.stock >= item.quantity for item in instance.items.all())


def is_staff_member(instance, user, **kwargs):
    return user.is_staff


def reserve_stock(instance, **kwargs):
    for item in instance.items.all():
        item.product.stock -= item.quantity
        item.product.save()


def send_approval_email(instance, **kwargs):
    ...


class OrderProcess(Process):
    process_name = 'process'
    transitions = [
        Transition(
            action_name='approve',
            sources=['draft'],
            target='approved',
            conditions=[has_stock],
            permissions=[is_staff_member],
            side_effects=[reserve_stock],
            callbacks=[send_approval_email],
        ),
        Transition(
            action_name='cancel',
            sources=['draft', 'approved'],
            target='cancelled',
        ),
    ]

Each declaration slot has one job:

• conditions — functions that answer True or False. Every one must answer True, or the transition is not available.
• permissions — functions that answer whether this user may run the transition. They receive user.
• side_effects — the work of the transition. It runs before the object reaches the target state. A failure stops the state change, and django-logic writes failed_state when you declare one.
• callbacks — functions that run after the object reaches the target state. They are best-effort: django-logic swallows what they raise.
• failure_callbacks — functions that run after a side-effect fails. They receive exception=. Put cleanup and compensation here.

Declare a Transition with no target for work that needs conditions, permissions and side-effects but writes no state on success. It follows the same rules as every transition: it takes the state lock, it is refused while a background transition is uncompleted, and it runs next_transition.

Add lock=False when the unit of work is not the bound row. A store process that posts one parcel's tracking is the example: the row is the store, the work is the parcel, and two parcels must not wait for each other. A lock=False transition takes no state lock and is not refused while a background transition is uncompleted. It keeps conditions, permissions, side_effects, callbacks, failure_callbacks and next_transition — a follow-up is best-effort, so one that takes the lock is skipped while the row is locked. It cannot declare a target or a failed_state — a state write must serialise on that state — and BackgroundTransition refuses it. get_available_actions() lists it while the row is locked, because it can be called.

def post_tracking(instance, **kwargs):
    ...  # tell the store about the parcel in kwargs['parcel']


Transition(
    action_name='post_tracking',
    sources=['approved', 'fulfilled'],
    lock=False,
    side_effects=[post_tracking],
)

permissions treat a call with no user as permitted. When one action name is split across nested processes by who is asking, give the person's branch permissions_class = StrictPermissions (refuses a call with no user) and the automation branch permissions_class = NoUserPermissions (refuses a call with one), both from django_logic.commands, so no caller matches two branches.

Bind the model to the process

Bind in your app's AppConfig.ready(). This is the one supported place.

# apps.py
from django.apps import AppConfig
from django_logic import ProcessManager


class ShopConfig(AppConfig):
    name = 'shop'

    def ready(self):
        from .models import Order
        from .process import OrderProcess
        ProcessManager.bind_model_process(Order, OrderProcess, state_field='status')

Import the model and the process inside ready(). A process references its model, and so do its condition, permission and side-effect functions. Binding at module import time therefore builds the import cycle models.py → process.py → actions.py → models.py. Django loads every app's models before it runs any ready(), so binding here cannot build that cycle, and your action modules import the model at the top level like normal code.

List the app in INSTALLED_APPS, or Django never runs ready().

One physical model and process name must identify one state field, including bindings through proxy models. Give independent fields distinct process names. Binding sibling proxies with the same process name to different fields is unsupported. The library does not validate that combination.

For an inherited state field on a multi-table model, drive the process through the parent model that owns the field. Mixing parent and child process access is unsupported. Their background identities and pending-work probes do not share one gate. Use a separate concrete model for an independent workflow.

Run a transition

order = Order.objects.get(pk=pk)
order.process.approve(user=request.user)

• The accessor is the process class's process_name. It is process by default.
• Pass user= in a request handler. A call without user= is a system call, and it skips every permission check.
• Never pass the request. A transition refuses request= at the call: hooks are fn(instance, **kwargs) and run on a worker for a background transition, where no request exists. Resolve what a hook needs in the view and pass plain values.
• order.process.get_available_actions(user=request.user) lists what this user may run right now.
• A refused transition raises TransitionNotAllowed from django_logic.exceptions. Its subclass TransitionTemporarilyUnavailable means the instance is busy, so the caller may retry shortly. Catch the subclass first.

The exception's reason is a RefusalReason from django_logic.exceptions. Its values are strings and can be included in JSON responses. Process calls also attach user_message, ready to display. Existing exception classes, diagnostic text from str(error), and exception arguments stay unchanged.

Reasons describe the existing checks; they do not change retry behavior. In particular, locked still raises plain TransitionNotAllowed. When declarations share an action name, prefer a transition whose sources include the current state and whose process permissions and conditions pass. Its failed check supplies the reason. Otherwise, use a process guard that blocks a declaration for the current state. If all declarations require another source state, report source_state.

Declaration order resolves ties within each group. A custom check that supplies no reason keeps None, even if a lower-priority refusal supplies a reason. This rule does not infer a unique intended branch or list every failed check. Process guards still run before their transitions and nested processes.

The resolver captures reasons during the existing checks. It does not run permissions or conditions again to explain a refusal. The existing available_actions hint still evaluates the available actions separately. A failed hint cannot replace the reason.

A custom is_valid override can return False without explaining why. Its reason is None unless a stock check on that same object recorded a failure. Caller-created exceptions also default to None. Continue to catch TransitionNotAllowed when handling these refusals.

Refusal messages

Process calls provide generic messages, such as Action 'submit' is not allowed: this record does not meet the requirements for this action. Pending background work uses is temporarily unavailable in the prefix. An unknown reason code or a custom check with no reason uses Action 'submit' is not allowed.

Use Process.refusal_messages for process-wide wording and the same option on a transition for its wording. These mappings contain overrides only. Use @refusal_message(text) for one condition or permission callable:

from django_logic import Process, Transition
from django_logic.commands import refusal_message
from django_logic.exceptions import RefusalReason, TransitionNotAllowed


@refusal_message("Upload an invoice before submitting.")
def has_invoice(claim):
    return claim.invoices.exists()


class ClaimProcess(Process):
    refusal_messages = {
        RefusalReason.PERMISSION: "Only the claim owner can perform this action.",
    }
    transitions = [
        Transition(
            "submit", sources=["draft"], target="submitted",
            conditions=[has_invoice],
            refusal_messages={
                RefusalReason.SOURCE_STATE: "This claim has already been submitted.",
            },
        ),
    ]


try:
    claim.process.submit(user=request.user)
except TransitionNotAllowed as error:
    response = {"reason": error.reason, "message": error.user_message}

Values are literal explanation text. The library adds the action prefix; it does not substitute templates or call message functions. The decorator returns the same callable with the same signature and behavior.

The first failed callable's message takes precedence over the transition map, then the selected nested process map, the root process map, and built-in defaults. A nested process with no override keeps the root's wording. A process-level failure uses process messages because no transition was selected. Only checks already evaluated can supply a message.

Message capture does not repeat checks or change short-circuit behavior. Nested refusals keep an already supplied user_message. Technical exceptions retain their normal behavior. Direct Transition.change_state() calls can leave user_message=None; process configuration applies to Process calls.

Background transitions

BackgroundTransition runs its side-effects on a worker process, and retries them. Declared with no target, it changes no state on success — same durability, same rules. Import it from django_logic.background.

# process.py
from django_logic import Process, Transition
from django_logic.background import BackgroundTransition


class OrderProcess(Process):
    transitions = [
        Transition(action_name='approve', sources=['draft'], target='approved'),
        BackgroundTransition(
            action_name='fulfil',
            sources=['approved'],
            target='fulfilled',
            in_progress_state='fulfilling',
            failed_state='fulfilment_failed',
            queue='django_logic.critical',
            timeout=600,
            side_effects=[book_courier, print_labels],
            callbacks=[send_tracking_email],
        ),
    ]

# views.py — returns as soon as django-logic saves the row.
order.process.fulfil(user=request.user)

The call writes in_progress_state and one TransitionMessage row in a single transaction, then returns. A worker claims that row, runs the side-effects and writes the target state, all in one atomic block. A failed attempt rolls back its own database writes and becomes claimable again after TRANSITION_MESSAGE_RETRY_MINUTES. After TRANSITION_MESSAGE_MAX_ERRORS attempts django-logic writes failed_state, runs failure_callbacks and completes the row.

• Side-effects must be idempotent against external systems. A retry runs them again from the start, so a payment or an email can happen twice.
• queue= is optional. A transition without it runs on DJANGO_LOGIC['DEFAULT_QUEUE'], which is django_logic. Name a queue per service level and give it its own worker.
• timeout= is optional. The worker kills an attempt that runs past it and records one error on the row.
• Raise PermanentFailure from django_logic.background when a retry cannot help. The worker then takes the failure path on the first attempt. For an exception type you do not own, list it in no_retry_on=(CustomsRefusal,).
• While a row is uncompleted, a second background transition on the same instance and process raises AlreadyInProgress, and a synchronous transition on it raises TransitionTemporarilyUnavailable. Start follow-up work from a callback, which runs after django-logic completes the row.

docs/design/PULL_WORKERS.md explains how a worker claims a row and what happens when one dies.

Run the workers

One dl_worker process serves one group of queues.

python manage.py dl_worker --queues django_logic.critical,django_logic.fast
python manage.py dl_worker --queues django_logic.slow --concurrency=4

--concurrency=N says how many attempts one worker runs at a time. The default is 1. Each attempt runs in its own forked process and holds its own database connection. Read "Sizing a deployment" in docs/design/PULL_WORKERS.md for the memory and connection budget.

A worker that dies releases its row lock with its database connection, and the next claim takes the row.

See what is not moving

python manage.py dl_transitions
python manage.py dl_transitions --send 1234

dl_transitions lists every uncompleted background transition and says why each one is not moving: it has spent every attempt, it waits out the retry pause, a worker runs it now, or no worker serves its queue. --queues narrows the list.

--send <pk> clears the retry wait on one row and wakes the workers, so the next claim takes it. The command runs no side-effects itself.

Two safety nets

The worker starts two safety nets once a minute in one separate process. Only one safety-net process runs per worker. Its time limit is 60 seconds. Nothing else needs a schedule.

• detect_stuck_transitions finalizes a row that has spent every attempt: it writes failed_state, runs failure_callbacks and completes the row. It also reports a row that waited past the retry window with no attempt, which means no worker serves that row's queue.
• cleanup_completed_transitions deletes completed rows older than TRANSITION_MESSAGE_CLEANUP_DAYS. It keeps the newest failed row per instance and process, because that row is the only explanation for an instance parked in its failed_state.

Alert when the worker processes stop. The safety nets stop with them. Reserve one extra database connection for the safety-net process, plus any connections your callbacks open. A slow or exiting callback cannot block the supervisor's attempt timeouts. If a pass stops, the next pass checks the remaining uncompleted rows. Callbacks for completed rows remain best-effort and are not retried.

Test your process

django_logic.testing gives you ProcessScenario, a test base class that runs a whole workflow inline, background transitions included. A test reads like the business process.

from django_logic.testing import ProcessScenario


class TestOrderFulfilment(ProcessScenario):
    process_class = OrderProcess
    model = Order
    state_field = 'status'

    def test_happy_path(self):
        order = self.create_instance(status='approved')
        self.background_transition(order, 'fulfil')
        self.assert_state(order, 'fulfilled')
        self.assert_side_effects_ran(['book_courier'])

Sync execution runs the worker path inline, for tests only. A test settings module calls django_logic.conf.enable_sync() and sets DJANGO_LOGIC['BACKGROUND_EXECUTION'] = 'sync'. Boot refuses that value anywhere else. docs/TESTING_GUIDE.md has the setup and the full scenario catalog.

More documentation

• docs/TESTING_GUIDE.md — how to test a process, and the assertions ProcessScenario gives you.
• docs/design/PULL_WORKERS.md — how the workers claim and run rows, how to size a deployment, and how to run behind pgbouncer.
• docs/recipes/nested-processes.md — how a parent drives many children without cascading their failures.
• docs/recipes/long-jobs.md — how to split work that runs for a long time.
• docs/logger.md — the loggers and the events they write.
• CHANGELOG.md — what each release changed, and every upgrade step.

Contributing

Pull requests are welcome. Open an issue first for a major change.

pip install -e '.[dev]'
python tests/manage.py test          # SQLite suite

make build                           # or run the same suite in Docker
make test
make test-one t=tests.test_transition

Add a test for every change, and update the documentation the change touches.

Report a bug in the issue tracker.

License

MIT