Metadata-Version: 2.4
Name: django-logic
Version: 0.14.1
Summary: Declarative business logic & state machines for Django — sync and durable background transitions
Author-email: Emil Balashov <emil@borderless360.com>, Andrey Omelyanuk <omelyanukandrey@gmail.com>
License: MIT License
Project-URL: Homepage, https://github.com/Borderless360/django-logic
Project-URL: Repository, https://github.com/Borderless360/django-logic
Keywords: django,fsm,state-machine,workflow,celery,background,business-logic
Classifier: Development Status :: 4 - Beta
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Intended Audience :: Developers
Classifier: Framework :: Django
Classifier: Framework :: Django :: 4.2
Classifier: Framework :: Django :: 5.1
Classifier: Framework :: Django :: 5.2
Classifier: Framework :: Django :: 6.0
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: django!=5.0.*,>=4.2
Requires-Dist: django-model-utils>=4.5.1
Requires-Dist: celery>=5.0
Provides-Extra: celery
Provides-Extra: redis
Requires-Dist: django-redis>=5.0.0; extra == "redis"
Provides-Extra: dev
Requires-Dist: coverage>=7.0; extra == "dev"
Requires-Dist: psycopg[binary]>=3.1; extra == "dev"
Requires-Dist: redis>=4.2; extra == "dev"
Dynamic: license-file

![django-logic](https://user-images.githubusercontent.com/6745569/87846635-dabb1500-c903-11ea-9fae-f1960dd2f82d.png)

[![CI](https://github.com/Borderless360/django-logic/actions/workflows/ci.yml/badge.svg)](https://github.com/Borderless360/django-logic/actions/workflows/ci.yml)
[![Coverage Status](https://coveralls.io/repos/github/Borderless360/django-logic/badge.svg?branch=master)](https://coveralls.io/github/Borderless360/django-logic?branch=master)
[![License](https://img.shields.io/pypi/l/django-logic.svg)](https://github.com/Borderless360/django-logic/blob/master/LICENSE)
     
Django Logic is a lightweight workflow framework for Django. It models business logic as a finite-state machine (FSM). You declare the state transitions, the permissions and the side effects in one place, away from your views, models and forms.

## Table of Contents
- [Features](#features)
- [Requirements](#requirements)
- [Installation](#installation)
- [Quick Start](#quick-start)
- [Core Concepts](#core-concepts)
- [Usage](#usage)
- [Complete Example](#complete-example)
- [Django-Logic vs Django FSM](#django-logic-vs-django-fsm)
- [Background Transitions](#background-transitions)
- [Testing Your Processes](#testing-your-processes)
- [Contributing](#contributing)
- [License](#license)

## Features
- 🎯 **Clear Business Logic** - Separate business logic from views, models, and forms
- 🔒 **Built-in Permissions** - Define who can perform which transitions
- 🔄 **Side Effects** - Execute functions during state transitions
- 🏗️ **Nested Processes** - Build complex workflows with sub-processes
- ⚡ **Built-in Locking** - Cache/Redis-based locking to prevent race conditions
- ⏳ **Durable Background Transitions** - Background transitions run as Celery tasks by default — built in, not an optional extra. They are queue-routed, retried, and recovered after a crash (see [Background Transitions](#background-transitions))
- 🧪 **Scenario-Based Testing** - Test a whole workflow as ordinary unit tests, background jobs, failures and retries included. Sync execution mode and `django_logic.testing` need no Celery broker (see [Testing Your Processes](#testing-your-processes))
- 🔍 **Structured Logging** - State changes go to the standard `django-logic` / `django-logic.transition` Python loggers. Configure them in Django `LOGGING` (see [docs/logger.md](docs/logger.md))

## Requirements
- Python 3.11+
- Django 4.2+ (4.2, 5.1, 5.2 and 6.0 are tested in CI; 5.0 is not supported)
- django-model-utils >= 4.5.1
- celery >= 5.0 — **installed automatically**; background transitions are Celery tasks
- A **cross-process `default` cache** for the state lock — *not* a package dependency. The engine locks through Django's cache API, so any cross-process backend works. That includes `django.core.cache.backends.redis.RedisCache`, built into Django since 4.0. Celery mode refuses to boot on a locmem/dummy cache when `DEBUG=False`.

Extras:
- `pip install django-logic[redis]` — installs `django-redis`, for deployments whose settings name `django_redis.cache.RedisCache`. It stopped being a core dependency in 0.11.0, because the engine has never imported it.
- `[celery]` remains an **empty alias**, so existing `pip install django-logic[celery,redis]` pins keep resolving — celery is a core dependency since 0.4

## Installation

```bash
# Installs the current release from PyPI. Celery is installed automatically.
# Add [redis] if your settings name django_redis.cache.RedisCache.
pip install django-logic
```

This README documents the current release line (the API introduced in 0.4).
The API changed a lot in 0.2–0.4. If you upgrade from a release before 0.2 — the
old DRF/Celery-coupled API — read [CHANGELOG.md](CHANGELOG.md) first. It lists
every breaking change and the steps to migrate.

## Quick Start

Here's a simple example to get you started:

```python
# models.py
from django.db import models

class Order(models.Model):
    STATUS_CHOICES = [
        ('pending', 'Pending'),
        ('paid', 'Paid'),
        ('shipped', 'Shipped'),
        ('delivered', 'Delivered'),
        ('cancelled', 'Cancelled'),
    ]
    status = models.CharField(max_length=16, choices=STATUS_CHOICES, default='pending')
    # ... other fields

# process.py
from django_logic import Process, Transition

class OrderProcess(Process):
    transitions = [
        Transition(
            action_name='pay',
            sources=['pending'],
            target='paid'
        ),
        Transition(
            action_name='ship',
            sources=['paid'],
            target='shipped'
        ),
        Transition(
            action_name='deliver',
            sources=['shipped'],
            target='delivered'
        ),
        Transition(
            action_name='cancel',
            sources=['pending', 'paid'],
            target='cancelled'
        ),
    ]

# apps.py — bind the process in your app's AppConfig.ready(). This is the only
# supported place to bind. ready() runs after every app's models are loaded, so
# binding here cannot create the model -> process -> actions -> model circular
# import that binding in models.py or process.py creates. See "Bind the
# process" below.
from django.apps import AppConfig
from django_logic import ProcessManager

class OrdersConfig(AppConfig):
    name = 'orders'

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

# Usage
order = Order.objects.create()
order.process.pay()  # Changes status from 'pending' to 'paid'
```

## Core Concepts

### Definitions
- **Transition** - Changes the state of an object from one state to another. It holds conditions, permissions, side-effects, callbacks and failure callbacks.
- **Action** - Like a transition, but it does not change the state. Use it for work that needs permissions and side-effects only.
- **Side-effects** - Functions that run during a transition, before the object reaches the target state. If one of them fails the state does not advance, and django-logic writes `failed_state` when you declare one. A failed background attempt also rolls back the database writes it made (a savepoint). A synchronous side-effect's writes stay.
- **Callbacks** - Functions that run after the object reaches the target state.
- **Failure callbacks** - Functions that run after a side-effect fails. django-logic writes `failed_state` and unlocks the state first. Put cleanup and compensation here.
- **Conditions** - Functions that must return True for a transition to be allowed.
- **Permissions** - Functions that check if a user can perform a transition.
- **Process** - Groups related transitions with common conditions and permissions.

## Usage
### 1. Add to INSTALLED_APPS
```python
INSTALLED_APPS = (
    ...
    'django_logic',
    ...
)
```

### 2. Define django model with one or more state fields
```python
from django.db import models


MY_STATE_CHOICES = (
     ('draft', 'Draft'),
     ('approved', 'Approved'),
     ('paid', 'Paid'),
     ('void', 'Void'),
 )

class Invoice(models.Model):
    my_state = models.CharField(choices=MY_STATE_CHOICES, default='draft', max_length=16, blank=True)    
    my_status = models.CharField(choices=MY_STATE_CHOICES, default='draft', max_length=16, blank=True)
    is_available = models.BooleanField(default=True)
    
```

### 3. Define a process class with some transitions
```python
from django_logic import Process as BaseProcess, Transition, Action
from .models import MY_STATE_CHOICES


# Define your side effect functions
def update_data(instance, **kwargs):
    # Update instance data
    for key, value in kwargs.items():
        if hasattr(instance, key):
            setattr(instance, key, value)
    instance.save()

class MyProcess(BaseProcess):
    transitions = [
        Transition(action_name='approve', sources=['draft'], target='approved'),
        Transition(action_name='pay', sources=['approved'], target='paid'),
        Transition(action_name='void', sources=['draft', 'approved'], target='void'),
        # An Action runs side-effects without changing state. `sources` lists
        # the states it's available from (required — there is no wildcard).
        Action(action_name='update', sources=['draft', 'approved'], side_effects=[update_data]),
    ]
```

### 4. Bind the process in your app's `AppConfig.ready()`

**Binding happens in exactly one place: your app's `AppConfig.ready()`.** Do
**not** bind at module import time (in `models.py` or `process.py`).

A process references its model, and so do its side-effect, condition and
permission functions. Binding `Model ⇄ Process` at import time therefore forces
`models.py → process.py → actions.py → models.py` — a circular import. The only
way out is a `from .models import X` line inside every action function.
`ready()` removes the cycle instead. Django imports **all** apps' models before
it runs **any** `ready()`, so every model already exists when you bind, and your
action modules can import the model at the top level like normal code.

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


class InvoicingConfig(AppConfig):
    name = 'invoicing'

    def ready(self):
        # Import inside ready() — never at module top in apps.py.
        from .models import Invoice
        from .process import MyProcess
        ProcessManager.bind_model_process(Invoice, MyProcess, state_field='my_state')
```

Then drive it from request/task/method bodies via `invoice.process.<action>(...)`
— never at module-import time or in another app's `ready()`.

> Make sure the app is wired so `ready()` runs — list it in `INSTALLED_APPS`
> (Django auto-discovers the single `AppConfig` in `apps.py`).


### 5. Advance your process with conditions, side-effects, and callbacks
Use next_transition to automatically continue the process.

> `next_transition` chains from a completed state change. A synchronous `Action`
> changes no state, so it ignores `next_transition`. A `BackgroundAction` does
> run it, from the worker. For a synchronous `Action`, start the follow-up from a
> callback instead, or use a `Transition`.

```python 
# Define permission and condition functions
def is_accountant(instance, user):
    return user.groups.filter(name='accountants').exists()

def is_customer_active(instance):
    return instance.customer.is_active if hasattr(instance, 'customer') else True

def generate_pdf_invoice(instance, **kwargs):
    # Generate PDF logic here
    pass

def send_approved_invoice_email_to_accountant(instance, **kwargs):
    # Send email logic here
    pass

def make_payment(instance, **kwargs):
    # Payment processing logic here
    pass

def send_void_invoice_email_to_accountant(instance, **kwargs):
    # Send void notification logic here
    pass

class MyProcess(BaseProcess):
    process_name = 'my_process' 
    permissions = [
        is_accountant, 
    ]
    transitions = [
        Transition(
            action_name='approve',
            sources=['draft'], 
            target='approved',
            conditions=[
                is_customer_active, 
            ],
            side_effects=[
                generate_pdf_invoice, 
            ],
            callbacks=[
                send_approved_invoice_email_to_accountant, 
            ],
            next_transition='pay' 
        ),
        Transition(
            action_name='pay',
            sources=['approved'],
            target='paid',
            side_effects=[
                make_payment, 
            ]
        ),         
        Transition(
            action_name='void', 
            callbacks=[
                send_void_invoice_email_to_accountant
            ],
            sources=['approved'],
            target='void'
        ),
        Action(
            action_name='update', 
            sources=['draft', 'approved'],
            side_effects=[
                update_data
            ],
        ),
    ]
```

### 6. Business logic explanation
This approval process holds these business rules:
- The user who performs the action must have the accountant role (permission).
- You cannot invoice an inactive customer (condition).
- When the invoice is approved, generate a PDF file and email it to an
accountant (side-effects and callbacks).
- When the invoice is voided, notify the accountant.

These business rules do not know about each other. You can also test every
function on its own, because Django-Logic connects them into the business
process for you.

### 7. Declarations are specifications

A process declaration is read by the person asking "how does this process
behave?" — not "how is this code structured?". Write it for that reader:

- **Write every process and every transition out in full.** Explicit
  `sources`, `target`, `conditions`, `side_effects`, `callbacks` per
  declaration — even when sibling processes repeat the same shape.
- **Duplication in declarations is acceptable, and usually preferable.**
  That is abnormal for ordinary code, and deliberate here: repeated
  declarations tell how the *process* works; a builder that assembles
  transitions tells how the *code* works, and hides the one line the
  reader came to check.
- **The same rule applies where the process is used.** Code that runs a
  process reads `instance.process.approve(...)` literally — not
  `getattr(process, action_name)` with the name held in a variable, and
  not a wrapper that hides which transition runs.

```python
# Hard to review: what are refund's sources? Which hooks run on cancel?
transitions = [_build_money_transition(name, hooks) for name, hooks in TABLE]

# Reviewable: each behaviour is stated where the reader looks for it.
transitions = [
    Transition(action_name='refund', sources=['paid'], target='refunded',
               side_effects=[reverse_charge], callbacks=[notify_customer]),
    Transition(action_name='cancel', sources=['draft', 'approved'], target='cancelled',
               side_effects=[release_stock]),
]
```

### 8. Execute in the code
```python
from invoices.models import Invoice


def approve_view(request, pk):
    invoice = Invoice.objects.get(pk=pk)
    # Check available transitions
    available_actions = invoice.my_process.get_available_actions(user=request.user)
    
    if 'approve' in available_actions:
        invoice.my_process.approve(user=request.user, context={'my_var': 1})
```
Use context to pass data between side-effects and callbacks.

> ⚠️ **django-logic checks permissions only when you pass `user=`.** A call
> without it (`invoice.my_process.approve()`) is a *system call*, and it
> **skips every permission check** by design. That is what you want in a Celery
> task or a management command. It is dangerous when you forget it in an API
> view. In a request handler, always pass `user=request.user`.

### 9. Handle state field overrides
If you want to override the value of the state field, it must be done explicitly. For example: 
```python
Invoice.objects.filter(my_state='draft').update(my_state='approved')
# or 
invoice = Invoice.objects.get(pk=pk)
invoice.my_state = 'approved'
invoice.save(update_fields=['my_state'])
```
When you change the state field by hand, always pass `update_fields=['my_state']`, as shown above. django-logic writes the state the same way. A transition therefore touches only the state column, and it never overwrites a field that a side-effect changed. Follow the same pattern in your own code. A plain `instance.save()` still saves the field like any other field — django-logic does not intercept it.

### 10. Error handling
```python 
from django_logic.exceptions import TransitionNotAllowed

try:
    invoice.my_process.approve()
except TransitionNotAllowed as e:
    logger.error(f'Approve is not allowed: {e}') 
```

## Complete Example

Here's a complete working example of an order processing system:

```python
# models.py
from django.db import models
from django.contrib.auth.models import User

class Order(models.Model):
    STATUS_CHOICES = [
        ('draft', 'Draft'),
        ('pending', 'Pending Payment'),
        ('paid', 'Paid'),
        ('processing', 'Processing'),
        ('shipped', 'Shipped'),
        ('delivered', 'Delivered'),
        ('cancelled', 'Cancelled'),
        ('refunded', 'Refunded'),
    ]
    
    status = models.CharField(max_length=16, choices=STATUS_CHOICES, default='draft')
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)
    is_paid = models.BooleanField(default=False)
    shipping_address = models.TextField(blank=True)
    tracking_number = models.CharField(max_length=100, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

class Product(models.Model):
    name = models.CharField(max_length=100)
    stock = models.PositiveIntegerField(default=0)

class OrderItem(models.Model):
    order = models.ForeignKey(Order, related_name='items', on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.PROTECT)
    quantity = models.PositiveIntegerField(default=1)

# conditions.py
#
# A condition that raises propagates out of the action call and out of
# get_available_actions(). Keep conditions cheap: read fields that always
# exist, and return False for a missing value instead of raising.
def has_stock_available(instance):
    return all(item.product.stock >= item.quantity for item in instance.items.all())

def is_payment_verified(instance):
    return instance.is_paid

def has_shipping_address(instance):
    return bool(instance.shipping_address)

# permissions.py
def is_customer(instance, user):
    return instance.user == user

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

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

def process_payment(instance, **kwargs):
    # Payment gateway integration
    instance.is_paid = True
    instance.save()

def generate_tracking_number(instance, **kwargs):
    import uuid
    instance.tracking_number = f"TRACK-{uuid.uuid4().hex[:8].upper()}"
    instance.save()

def send_order_confirmation_email(instance, **kwargs):
    # Send email to customer
    pass

def send_shipping_notification(instance, **kwargs):
    # Send tracking info to customer
    pass

# process.py
from django_logic import Process, Transition

from .conditions import has_stock_available, has_shipping_address, is_payment_verified
from .permissions import is_customer, is_staff_member
from .side_effects import (
    generate_tracking_number,
    process_payment,
    reserve_stock,
    send_order_confirmation_email,
    send_shipping_notification,
)

class OrderProcess(Process):
    process_name = 'order_process'
    
    transitions = [
        Transition(
            action_name='submit',
            sources=['draft'],
            target='pending',
            conditions=[has_stock_available, has_shipping_address],
            side_effects=[reserve_stock],
        ),
        Transition(
            action_name='pay',
            sources=['pending'],
            target='paid',
            side_effects=[process_payment],
            callbacks=[send_order_confirmation_email],
            next_transition='process',
        ),
        # The automatic follow-up that 'pay' chains into. It declares no
        # permission on purpose. next_transition forwards the original
        # caller's user, so a staff-only permission here would stop the chain
        # for the customer who paid, and django-logic swallows a rejected
        # follow-up. Its condition is the payment check that 'pay' has just
        # satisfied.
        Transition(
            action_name='process',
            sources=['paid'],
            target='processing',
            conditions=[is_payment_verified],
        ),
        Transition(
            action_name='ship',
            sources=['processing'],
            target='shipped',
            permissions=[is_staff_member],
            side_effects=[generate_tracking_number],
            callbacks=[send_shipping_notification],
        ),
        Transition(
            action_name='deliver',
            sources=['shipped'],
            target='delivered',
        ),
        Transition(
            action_name='cancel',
            sources=['draft', 'pending'],
            target='cancelled',
            permissions=[is_customer],
        ),
        Transition(
            action_name='refund',
            sources=['paid', 'processing', 'shipped', 'delivered'],
            target='refunded',
            permissions=[is_staff_member],
        ),
    ]

# apps.py — bind in AppConfig.ready() (the one supported place; see "Bind the
# process"). Never bind at module import time.
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')

# views.py
from django.shortcuts import render, redirect
from django.contrib import messages
from django_logic.exceptions import TransitionNotAllowed

from .models import Order

def submit_order(request, order_id):
    order = Order.objects.get(pk=order_id, user=request.user)
    
    try:
        order.order_process.submit(user=request.user)
        messages.success(request, 'Order submitted successfully!')
    except TransitionNotAllowed as e:
        messages.error(request, f'Cannot submit order: {str(e)}')
    
    return redirect('order_detail', order_id=order.id)
```

## Troubleshooting

### Common Issues

#### 1. TransitionNotAllowed Exception
django-logic raises this exception for two different reasons. A generic handler
should answer them differently:

**Permanent refusal** — retrying is pointless:
- The current state is not in the transition's source states
- Conditions are not met
- User doesn't have required permissions
- State is already locked by another process

**Transient concurrency** — the action is fine, retry shortly. These raise
`TransitionTemporarilyUnavailable`, a `TransitionNotAllowed` subclass you can
import from `django_logic.exceptions`. You get it in two cases: another
transition owns the instance right now (`AlreadyInProgress`, or the synchronous
gate while a background row is still being retried), or the state moved while
enqueue waited (`SourceStateChanged`). A stranded row — nothing is retrying it —
does not count as busy. The synchronous gate and a later enqueue then raise the
plain base class again, so "retry shortly" is never a forever answer. An attempt
that still runs inside its declared `timeout=` budget always counts as being
retried. Catch the transient type **ahead of** the base class:

```python
from django_logic.exceptions import (
    TransitionNotAllowed,
    TransitionTemporarilyUnavailable,
)

try:
    order.process.submit(user=request.user)
except TransitionTemporarilyUnavailable:
    return Response(status=409, data={'detail': 'Busy — please retry shortly.'})
except TransitionNotAllowed:
    return Response(status=400, data={'detail': 'Action not allowed.'})
```

**Solution**: Check available transitions using `get_available_actions()` before calling a transition, and handle the transient subclass separately.

#### 2. State Not Updating
If the state field is not updating:
- Ensure you're not using `save()` without `update_fields`
- Check if the transition completed successfully
- Verify side effects didn't raise exceptions

**Solution**: Always use `update_fields=['state_field_name']` when manually saving state changes.

#### 3. Race Conditions
Multiple processes trying to transition the same object can cause race conditions.

**Solution**: Django-Logic serializes work on a state field with two mechanisms (see [Concurrency and locking](#concurrency-and-locking)):
- a **cache lock** — an atomic set-if-absent on the `default` cache. It is held for a synchronous transition's whole run, and for the critical section of a background transition's enqueue. Both re-read the persisted state under the lock.
- the **`TransitionMessage` row**. While a background transition is in progress, a second one raises `AlreadyInProgress`, and a synchronous transition on the same instance + process raises `TransitionTemporarilyUnavailable`. Both are `TransitionNotAllowed` subclasses.

Use a cross-process cache, so that the web processes and the workers share the lock.

#### 4. Side Effects Not Rolling Back
A side-effect that changes an external system does not roll back on its own.

**Solution**: Compensate in failure callbacks. They run after django-logic unlocks the state, so make them idempotent — another process can act on the instance in between:

```python
def compensate_payment(instance, exception, **kwargs):
    # Reverse the payment if a side-effect failed
    pass

Transition(
    action_name='pay',
    sources=['pending'],
    target='paid',
    side_effects=[process_payment, another_side_effect],
    failure_callbacks=[compensate_payment, notify_admin],   # run after unlock
)
```

When a side-effect fails, django-logic runs these steps in order: write `failed_state` (when you declare one), unlock, then run the **failure_callbacks**.

## Django-Logic vs Django FSM
[Django FSM](https://github.com/viewflow/django-fsm) came before Django-Logic.
Django-Logic removes some of its limits and adds new features:

### Key Differences:
- **Processes**: Django-Logic groups transitions into processes
- **Nested Processes**: Build workflows from sub-processes
- **Built-in Locking**: Prevents race conditions out of the box
- **Failure Handling**: Dedicated failure callbacks and failed states
- **Better Separation**: Keeps business logic apart from the code that runs it
- **Background Tasks**: Durable, queue-routed background execution built into `django_logic.background` ([Background Transitions](#background-transitions)) — no extra package

### Migration from Django FSM:
When you migrate from Django FSM, these are the main changes:
1. Replace the `@transition` decorator with the `Transition` class
2. Move the transition logic into side-effects and callbacks
3. Group related transitions into Process classes
4. Bind each model to its process with `ProcessManager.bind_model_process(...)` in your app's `AppConfig.ready()` (see [Bind the process](#4-bind-the-process-in-your-apps-appconfigready))

## Advanced Features

### Nested Processes
Build complex workflows by combining processes:

```python
class PaymentProcess(Process):
    transitions = [
        Transition('validate', sources=['pending'], target='validated'),
        Transition('charge', sources=['validated'], target='charged'),
    ]

class OrderProcess(Process):
    nested_processes = [PaymentProcess]
    transitions = [
        Transition('submit', sources=['draft'], target='pending'),
        # ... other transitions
    ]
```

### Custom State Classes
Extend the State class for custom behavior:

```python
from django_logic.state import State

class AuditedState(State):
    def set_state(self, state):
        # Log state changes
        audit_log.create(
            model=self.instance.__class__.__name__,
            instance_id=self.instance.pk,
            field=self.field_name,
            old_value=self.get_persisted_state(),
            new_value=state,
        )
        super().set_state(state)
```

Install it with `state_class` on the process — without this the subclass is never used:

```python
class OrderProcess(Process):
    state_class = AuditedState
    transitions = [...]
```

Every state read and write for that process then goes through your class. That includes the writes the background engine makes: `in_progress_state` at enqueue, and the target state and `failed_state` on the worker. Two rules apply. `get_persisted_state()` must keep reading the database row, because the re-read under the lock and the worker's state guard trust it — never return a cached value from it. And `set_state` must call `super()`.

### Context Passing
Pass data between side effects and callbacks:

> **Reserved kwarg names.** The engine sets `tr_id`, `root_id`, `parent_id`,
> `process_class` and `owning_process_class` on every run. They identify the
> transition and its parents, and the engine forwards them to a
> `next_transition` follow-up. The engine also replaces `user` with `user_id`
> when it sends the transition to the queue, and it rebuilds `context` on the
> worker. If you pass one of those names as your own data, the engine overwrites
> it. Use different names.

> `context` lives for **one execution** only; django-logic never saves it. A
> `context=` you pass reaches the hooks of a synchronous transition. For a
> *background* transition, enqueue drops it and the worker builds an empty one.
> It carries data between the hooks of one run, not across the queue. Anything
> the worker must see belongs in ordinary kwargs, which the engine serializes,
> or on the instance itself.


```python
def calculate_total(instance, context, **kwargs):
    total = sum(item.price for item in instance.items.all())
    context['total'] = total

def apply_discount(instance, context, **kwargs):
    total = context.get('total', 0)
    instance.final_amount = total * 0.9  # 10% discount
    instance.save()

Transition(
    action_name='checkout',
    sources=['cart'],
    target='pending',
    side_effects=[calculate_total, apply_discount],
)
```

## Background Transitions

For long-running side-effects (payment processing, PDF generation, external API calls), use `BackgroundTransition` / `BackgroundAction` from `django_logic.background`. **Background transitions are Celery tasks** — Celery ships as a core dependency and `'celery'` is the default execution mode.

**How the work is split.** A synchronous `Transition` does everything at once, in the caller's call frame. A background transition cannot: its work runs later, on another machine. It therefore follows the standard transactional-outbox pattern, in two steps:

- **Enqueue** (synchronous, inside your request): validate the transition, then write `in_progress_state` and a durable `TransitionMessage` row in **one** database transaction. The row records what you asked for. django-logic sends the Celery task on commit. This takes milliseconds.
- **Execute** (on a Celery worker): load the row, run the side-effects, write the target state and mark the row completed — all in one atomic block. The durable row is what lets the safety-net tasks retry or finalize the work when the worker crashes or the broker loses the message. Success and failure *callbacks* run after the worker's transaction commits, and they are best-effort by contract.

They give you:

- **Durable execution.** django-logic saves every background transition as a `TransitionMessage` row, in the same atomic block that writes `in_progress_state`. A periodic safety-net task recovers the work after a worker crash, a broker loss, or a dropped `transaction.on_commit` hook.
- **Queue routing per transition.** `queue=` is optional — a transition without it runs on `DJANGO_LOGIC['DEFAULT_QUEUE']` (`'django_logic'`). Name your queues per SLA (`critical` / `slow` / `fast`) and give each one its own worker.
- **Sync mode for tests.** `'sync'` runs the worker path inline, in the same process — for unit tests, CI, management commands and the Django shell. You need no Celery broker to test a business process; see [Testing Your Processes](#testing-your-processes).
- **One task per attempt, all or nothing.** Every side-effect and the target-state write happen inside **one** Celery task with `acks_late=True`, inside **one** atomic block, with the side-effects in a savepoint. A failed attempt **rolls back every database write it made**. A worker crash re-delivers the whole task, so the state never stops between two side-effects. The idempotency you owe is for *external* calls only, because a retried attempt runs the side-effects again from the start.

### Install

Add `'django_logic.background'` to `INSTALLED_APPS` and configure:

```python
DJANGO_LOGIC = {
    'LOCK_TIMEOUT': 7200,   # the state lock's TTL, seconds
    'BACKGROUND_EXECUTION': 'celery',   # the default; set 'sync' in test settings
    'DEFAULT_QUEUE': 'django_logic',    # queue for transitions without queue=
    'STARTER_QUEUE': 'django_logic.starter',
    'TRANSITION_MESSAGE_MAX_ERRORS': 5,
    'TRANSITION_MESSAGE_RETRY_MINUTES': 2,
    'TRANSITION_MESSAGE_CLEANUP_DAYS': 7,
    'STRICT_KWARGS_SERIALIZATION': False,  # True: raise (not warn) on dropped 'request' / non-string dict keys
    'STRICT_HOOK_SIGNATURES': False,    # True: refuse to bind hooks without a named instance-first parameter
    'DEFER_UNLOCK_UNTIL_COMMIT': False,  # True: sync unlocks ride transaction.on_commit (see "Concurrency and locking")
    # 'LEGACY_EXCEPTION_BASE': '...',  # opt-in: dotted path of a fork's TransitionNotAllowed to mix in during a migration (see below)
}
```

Every key has the default shown above, so an empty `DJANGO_LOGIC = {}` is a valid production start. `manage.py check` reports a key the engine does not read as `django_logic.W004` instead of ignoring it — that catches a typo and a key a past release removed. django-logic also validates the numeric and safety-critical settings at boot: a bad value raises `ImproperlyConfigured` and names the setting, so you find out before a worker fails on it. Run `manage.py migrate` to create the `TransitionMessage` table.

#### Migrating off a fork: `LEGACY_EXCEPTION_BASE`

A project that migrates off a fork of this library runs both engines side by
side while its apps move one at a time. Its shared handlers — DRF mixins, admin
actions, Sentry ignore lists — catch the *fork's* `TransitionNotAllowed`. Name
the fork's class here, and this engine's refusals become instances of it too, so
those handlers keep working:

```python
DJANGO_LOGIC = {
    'LEGACY_EXCEPTION_BASE': 'old_fork.exceptions.TransitionNotAllowed',
}
```

django-logic adds the class to `TransitionNotAllowed`'s bases in
`AppConfig.ready()`, so every subclass is covered, including
`TransitionTemporarilyUnavailable`. The setting costs nothing when you leave it
out. Every way it can go wrong — a path that does not import, a class that is
not an exception, an MRO conflict — raises `ImproperlyConfigured` at boot.
Remove the setting when the migration is done.

At boot, celery mode refuses two settings that would break the guarantees without saying so. The first is a SQLite database for `TransitionMessage`, which has no `select_for_update(nowait)`. The second is a per-process `default` cache (locmem or dummy) when `DEBUG=False`, because the web processes and the workers must share the state lock:

```python
CACHES = {
    'default': {
        # Built into Django since 4.0; needs the `redis` client package.
        # `django_redis.cache.RedisCache` also works — `pip install django-logic[redis]`.
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': os.environ['REDIS_URL'],
    }
}
```

### Declare a background transition

```python
from django_logic import Process, Transition
from django_logic.background import BackgroundTransition, BackgroundAction


class OrderProcess(Process):
    transitions = [
        Transition(
            action_name='approve',
            sources=['draft'],
            target='approved',
            side_effects=[validate_order],
        ),
        BackgroundTransition(
            action_name='fulfil',
            sources=['approved'],
            target='fulfilled',
            in_progress_state='fulfilling',
            failed_state='fulfilment_failed',
            queue='django_logic.critical',     # explicit queue: dedicated worker, tight SLA
            side_effects=[reserve_stock, generate_labels, call_courier],
            callbacks=[send_confirmation_email],
        ),
        BackgroundTransition(
            action_name='generate_export',
            sources=['fulfilled'],
            target='exported',
            in_progress_state='exporting',
            failed_state='export_failed',
            queue='django_logic.slow',         # slow work, isolated worker
            side_effects=[build_csv, upload_to_s3],
        ),
        BackgroundAction(
            action_name='sync_inventory',
            sources=['fulfilled'],
            # no queue= — runs on DEFAULT_QUEUE ('django_logic')
            side_effects=[push_to_erp],
        ),
    ]


# apps.py — bind in AppConfig.ready() (the one supported place; see "Bind the process").
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')
```

### Call it

```python
# In a view — returns immediately (Celery mode) or after the worker completes (Sync mode).
tr_id = order.process.fulfil(user=request.user)
```

### Polymorphic routing with nested processes

Nested processes let several sub-processes share one `action_name`. A
**condition on the instance** then picks one of them at run time, so a generic
caller invokes one method and the right implementation runs. This works for
background transitions too: each integration keeps its durable work on its own
nested process, and callers never have to know which one.

```python
def is_gmail(conversation, **kw):  return conversation.source_integration == 'gmail'
def is_dummy(conversation, **kw):  return conversation.source_integration == 'dummy'

class GmailConversationProcess(Process):
    process_name = 'gmail_conversation'
    transitions = [
        BackgroundTransition(
            action_name='send_message_via_integration',
            sources=['open'], target='open',
            in_progress_state='gmail_sending',
            conditions=[is_gmail],
            side_effects=[send_via_gmail],
        ),
    ]

class DummyConversationProcess(Process):
    process_name = 'dummy_conversation'
    transitions = [
        BackgroundTransition(
            action_name='send_message_via_integration',   # same name, different owner
            sources=['open'], target='open',
            in_progress_state='dummy_sending',
            conditions=[is_dummy],
            side_effects=[send_via_dummy],
        ),
    ]

class ConversationProcess(Process):
    nested_processes = [GmailConversationProcess, DummyConversationProcess]

# apps.py — bind in AppConfig.ready() (the one supported place; see "Bind the process").
from django.apps import AppConfig
from django_logic import ProcessManager

class MessagingConfig(AppConfig):
    name = 'messaging'

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

# Generic caller — routes by source_integration, no integration knowledge here:
conversation.process.send_message_via_integration(user=request.user)
```

Enqueue resolves exactly one transition, because the conditions exclude each
other. It records the **nested process class that declared the transition** on
the `TransitionMessage`. The worker restores that exact transition from the
recorded class and does not evaluate the condition again, so the routing holds
even when the instance changes while the row waits. One constraint: a background
`action_name` must be **unique within a single process class**, because the
worker cannot tell two transitions of the same name in one class apart. You may
share `in_progress_state` freely, which helps when a UI knows only one "busy"
value. Every marked instance carries its exact transition on the
`TransitionMessage` row, so recovery never has to guess which transition owns it.
(0.12.0 retired the `django_logic.E001` check that used to forbid sharing, along
with the stranded sweep.) A background `action_name` may also match a synchronous
transition of the same name: the worker restores background transitions only, and
enqueue routes the call by condition. A synchronous fast path and a durable
background slow path can therefore share one `action_name`.

> **`in_progress_state` is background-only (0.12.0).** On a `BackgroundTransition`
> django-logic writes the state in the same transaction as the
> `TransitionMessage` row, so a row owns the recovery of every marked instance. A
> *synchronous* transition used to write the state under a cache lock with no
> durable row. A killed process then left the instance parked in a state with no
> way out, and the engine needed a whole sweeping subsystem
> (`recover_stranded_states`, now retired) to find those instances. Declaring the
> state on a plain `Transition`/`Action` now raises `ImproperlyConfigured`.
> Without that write, a killed synchronous run rolls back to its source state and
> you can run it again once the lock TTL expires. There is nothing to sweep.
>
> **Migrating a sync transition that used the marker:** model the busy step as
> a real state with explicit edges —
>
> ```python
> Transition('submit', sources=['draft'], target='fulfilling',
>            next_transition='do_fulfil'),
> BackgroundTransition('do_fulfil', sources=['fulfilling'], target='fulfilled',
>                      failed_state='fulfilment_failed', side_effects=[...]),
> ```
>
> Readers still see `fulfilling`, and a `TransitionMessage` row now owns the
> work. The pattern has one narrow window that the old atomic write did not have.
> A crash between `submit`'s commit and the chained dispatch parks the instance at
> `fulfilling` with no row. A three-line periodic retry recovers it, and the retry
> is safe: an instance that is really in progress raises `AlreadyInProgress` and
> is skipped, while a parked one moves *forward*:
>
> ```python
> for obj in Order.objects.filter(status='fulfilling',
>                                 modified__lt=now() - timedelta(hours=1)):
>     try:
>         obj.process.do_fulfil()
>     except TransitionNotAllowed:
>         pass  # in progress or locked — someone owns it
> ```

> **Upgrade note.** You may turn an existing background transition with a unique
> name into this shared-name nested pattern. Deploy that change with no
> uncompleted rows for the action, or split it across two deploys. A row that
> older code enqueued does not name the process that declared the transition.
> Once several nested processes share the name, the worker cannot tell which one
> such a row meant, so it completes the row without running the side-effects. That
> is safe, but the work does not run. Every row enqueued after the upgrade records
> the process that declared it.

### Testing background transitions

Set `BACKGROUND_EXECUTION='sync'` in your test settings. The global default is `'celery'`, so you must opt in. Every `instance.process.fulfil(...)` call then enqueues **and** executes inline, with no broker:

```python
class FulfilmentTests(TestCase):
    def test_happy_path(self):
        order = Order.objects.create(status='approved')
        order.process.fulfil()
        order.refresh_from_db()
        self.assertEqual(order.status, 'fulfilled')

    def test_side_effect_failure_propagates(self):
        # Patch what the side-effect CALLS, not the side-effect itself. The
        # Transition captured the function object when the class was defined,
        # so patching its module attribute does not replace it.
        # django_logic.testing's fail_side_effect= avoids this trap.
        order = Order.objects.create(status='approved')
        with patch('myapp.services.courier_client.book', side_effect=CourierError):
            with self.assertRaises(CourierError):
                order.process.fulfil()
```

If the global setting is `'celery'` but you need Sync mode for a specific block, use the context manager:

```python
from django_logic.background import sync_execution

with sync_execution():
    order.process.fulfil()
```

### Suggested queue layout

```
django_logic.fast       — < 1s work (notifications, cache invalidations)
django_logic.critical   — user-facing with SLA (fulfilment, payments)
django_logic.slow       — > 30s work (exports, reports)
django_logic.starter    — the framework's periodic safety-net tasks
```

The periodic starter sends a stale transition to its own queue again, so a retried slow job never moves to the critical queue.

### Safety-net tasks

Four periodic tasks keep the durable model working on its own. Run them on `STARTER_QUEUE` through Celery beat:

- `retry_stale_transitions` — sends uncompleted rows older than `RETRY_MINUTES` to the queue again. It skips a row whose current attempt started less than `RETRY_MINUTES` ago, so a running attempt is not re-dispatched on every tick.
- `cleanup_completed_transitions` — deletes completed rows older than `CLEANUP_DAYS`.
- `detect_stuck_transitions` — finalizes a row that sits at `MAX_ERRORS`: it writes `failed_state`, runs `failure_callbacks` and marks the row completed, so the retry loop stops.
- `watchdog_stale_attempts` — gives up on an attempt that ran past its declared `timeout` (see below).

### Per-attempt timeouts

A `BackgroundTransition` (or a `BackgroundAction`) may give each attempt a wall-clock budget with `timeout=<seconds>`:

```python
BackgroundTransition(
    action_name='generate_export',
    sources=['fulfilled'],
    target='exported',
    in_progress_state='exporting',
    failed_state='export_failed',
    queue='django_logic.slow',
    timeout=600,                       # give up on an attempt after 10 minutes
    side_effects=[build_csv, upload_to_s3],
)
```

`watchdog_stale_attempts` looks for an uncompleted row whose current attempt (`started_at`) has run past `timeout`. It records a `TimeoutError` as a failed attempt. Once `errors_count` reaches `MAX_ERRORS`, it finalizes the row to `failed_state`. The watchdog ignores a row without `timeout`. It counts one error per attempt at most: if the attempt has already recorded an error of its own since it started, the watchdog leaves it alone. django-logic writes `started_at` in its own committed statement before the attempt begins. The value therefore stays visible while the attempt runs and survives a worker that dies, which is what makes a hung or crashed attempt visible at all. The watchdog cannot tell a crashed attempt from a slow one, so a re-dispatched attempt may run the side-effects again while the first attempt still runs. **Side-effects must be idempotent against external systems.** Their database writes are atomic per attempt and roll back on failure, but an external API call that both attempts make happens twice.

### Concurrency and locking

Two mechanisms serialize work on a state field, each with its own scope:

1. **The cache lock** — an atomic set-if-absent on the `default` cache. A *synchronous* transition holds it for its whole run. A background transition holds it for the **critical section of enqueue only**: validate, create the `TransitionMessage`, write `in_progress_state`, release. Both re-read the **persisted** state under the lock before they go on, so two requests that race on the same instance cannot both win.
2. **The uncompleted `TransitionMessage` row** gates concurrent background work. While one exists for an instance + process:
   - a second background transition raises `AlreadyInProgress` (`from django_logic.background.exceptions import AlreadyInProgress`). You can also catch it as `TransitionTemporarilyUnavailable` from `django_logic.exceptions`, without importing the background subpackage. A partial unique constraint enforces this, so it holds across processes and machines.
   - a **synchronous transition on the same instance + process raises `TransitionTemporarilyUnavailable`**, because the worker owns the state field until the row completes.
   - a synchronous `Action` still runs, because it changes no state on success. django-logic skips a failing Action's `failed_state` write while the row is uncompleted, for the same reason.

The constraint is scoped **per process**. Two independent state machines bound to different fields of the same model — say `status` and `payment_status` — can both have background work in progress.

**Data-dependent outcomes ("verdicts").** A background side effect cannot run a synchronous transition on its own instance: its own row is still uncompleted, so the gate above refuses it. When the side effect's result decides the next state (a validation that ends valid or invalid), write the decided call explicitly and run it from the transition's `callbacks`, which execute after the row completes:

```python
from functools import partial

def fetch_sheet(import_obj, **kwargs):
    rows = read_sheet(import_obj)
    if all(row.valid for row in rows):
        import_obj.run_verdict = partial(import_obj.process.mark_as_valid, data=rows)
    else:
        import_obj.run_verdict = partial(import_obj.process.mark_as_invalid,
                                         errors=errors_of(rows))

def apply_verdict(import_obj, **kwargs):
    run_verdict = import_obj.__dict__.pop('run_verdict', None)
    if run_verdict is not None:
        run_verdict()

BackgroundAction(action_name='run_validation', sources=['validating'],
                 side_effects=[fetch_sheet], callbacks=[apply_verdict])
```

Each decision point names the exact transition it runs. Callbacks are best-effort: a worker that dies between completing the row and the callback loses the verdict. The instance stays in its current state, and the operator can run the action again — nothing is corrupted.

To answer "busy, try again shortly" in your own API, read the row through `in_flight()` instead of writing the filter yourself:

```python
from django_logic.background import in_flight

if in_flight(order, 'process'):
    return Response(status=409, data={'detail': 'Busy — please retry shortly.'})
```

The answer can go out of date at once, because a transition can start or complete right after the read. Use it to shape an answer, not as a gate — the engine's own guards decide. It answers the *busy* question only: for a stranded row (uncompleted, nothing retrying it) it returns `False`, which matches the plain `TransitionNotAllowed` the engine's gates raise for such a row.

The gate is a database row, not a held lock, so nothing is left behind when the caller's surrounding transaction rolls back. The row, the `in_progress_state` write and the dispatch all disappear together.

**Lock ownership.** Every acquisition stores a unique token, and the release compares the token before it deletes the key. A synchronous run that outlives its lock TTL therefore cannot delete the lock that a later run acquired: the token does not match, so it leaves the lock alone and returns. A `State` object that never locked still deletes the key without a check, which gives you a way to release a lock by hand.

**Synchronous transitions inside an outer `transaction.atomic()`.** By default django-logic releases the lock as soon as the transition completes, before the outer block commits. That window is real. Another connection can take the lock, read the *old committed* state, and run the same transition again. Both runs then execute the side-effects, and the final state depends on which one commits last. Opt in when your code drives transitions inside atomic blocks and needs the exclusion to cover the whole uncommitted span:

```python
DJANGO_LOGIC = {..., 'DEFER_UNLOCK_UNTIL_COMMIT': True}
```

The unlock then runs from `transaction.on_commit`. Design for two trade-offs. On **rollback** the hook never runs, so the lock waits for its TTL to expire — a lockout with a known end, the same as after a crashed process. And a follow-up on the same instance (`callbacks` / `next_transition`) inside the atomic block finds the state still locked, so django-logic skips it; start those from `transaction.on_commit` in the caller instead. You can also keep the default and call the transition from `transaction.on_commit`, so it starts only once the surrounding write is visible.

One consequence: you **cannot** chain a background transition from another transition's `callbacks` or `next_transition` on the *same* instance while the first row is still uncompleted. The chained enqueue raises `AlreadyInProgress`. Start follow-up background work from a terminal hook — a success or failure callback that runs after django-logic marks the first row completed — or work on a different instance.

> ⚠️ **Do not treat `AlreadyInProgress` as "already queued, my changes will be picked up".** That is only true while the existing attempt has **not started**. If the worker already runs, it has already read its inputs. Your update lands after that read, the run commits a result computed from the older data, and nothing runs again. For a recompute-style transition, save a dirty flag (or a version) *before* you dispatch, clear it inside the side-effect, and dispatch again from a success callback when the flag is set again:
>
> ```python
> def recompute(instance, **kwargs):
>     Order.objects.filter(pk=instance.pk).update(recompute_requested=False)
>     ...  # compute from current rows
>
> def redispatch_if_dirty(instance, **kwargs):   # success callback (terminal)
>     instance.refresh_from_db()
>     if instance.recompute_requested:
>         instance.process.recompute_rates()
> ```

### The worker's state guard

The worker restores the transition by name and skips the source-state check on purpose. Something else can still move the instance while the row waits: an operator fix in the admin, a data migration, a support script. Retries span `RETRY_MINUTES × MAX_ERRORS`, so that collision does happen in production.

Before it runs the side-effects, the worker checks that the persisted state still matches what enqueue left behind — `in_progress_state`, or a declared source state when the transition has no `in_progress_state`. If it does not match, the worker completes the row as **superseded**: it skips the side-effects, the external state change wins, and it records the reason on the row (`last_error_message` starts with `[superseded]`) and logs it at ERROR.

The same check guards the `failed_state` writes that the safety-net tasks make, so a watchdog that finalizes a long-stranded row never overwrites a manual fix.

### Production deployment

Celery mode needs three things. Without them the durability guarantees do not hold, and nothing tells you:

**1. A real broker.** `BACKGROUND_EXECUTION='celery'` needs a durable broker (Redis or RabbitMQ). With no broker in the settings, Celery falls back to an in-memory transport that no worker reads. `apply_async` then succeeds and the task never runs. django-logic logs a warning once, on the first dispatch.

**2. The four periodic safety-net tasks, scheduled through Celery beat.** Celery registers them for you (`@shared_task`, names `django_logic.*`) once your Celery app imports or auto-discovers `django_logic.background.tasks`. **If you do not schedule them, you get no retries, no stuck-row finalization, no timeout watchdog and no cleanup.** One lost broker message then leaves an instance waiting forever.

Use the ready-made schedule. It routes all four tasks to `DJANGO_LOGIC['STARTER_QUEUE']`, and you can override each one by keyword. Retry runs every 60s, detect-stuck every 300s, and the watchdog every 120s. Cleanup runs at 03:17 on a crontab, not on an interval: an interval counts from the moment beat starts, so a platform that deploys or cycles dynos every day resets the clock before a day-scale interval ever elapses.

```python
# celery.py — after the app is configured
from django_logic.background import beat_schedule

# Write the CELERY_-namespaced key, not app.conf.beat_schedule. With
# config_from_object(namespace='CELERY'), Celery reads beat_schedule from
# CELERY_BEAT_SCHEDULE in Django settings first. It accepts a plain
# `app.conf.beat_schedule = ...` assignment and then ignores it.
app.conf['CELERY_BEAT_SCHEDULE'] = {
    **(app.conf.beat_schedule or {}),
    **beat_schedule(),
}
```

If the schedule does not reach beat, `manage.py check` reports the tasks as unscheduled (`django_logic.W002`).

A hand-written `CELERY_BEAT_SCHEDULE` works the same way. The task names are `django_logic.retry_stale_transitions`, `django_logic.detect_stuck_transitions`, `django_logic.watchdog_stale_attempts` and `django_logic.cleanup_completed_transitions`. Set `options={'queue': ...}` on each entry yourself.

Run a worker that consumes your transition queues **and** the starter queue, plus beat:

```bash
celery -A myproject worker -Q django_logic.critical,django_logic.slow,django_logic.fast,django_logic.starter
celery -A myproject beat        # (or `worker -B` in dev; run a single beat in production)
```

**3. Crash re-delivery is built in.** Every django-logic task sets
`acks_late=True` **and** `reject_on_worker_lost=True` on the task itself. The
broker therefore delivers the transition again when its worker dies while the
task runs — SIGKILL, OOM, a deploy, or a `--max-memory-per-child` kill — whatever
your global Celery configuration says. You wire up nothing. Setting the global
pair is still a good idea for your *own* tasks:

```python
CELERY_TASK_ACKS_LATE = True
CELERY_TASK_REJECT_ON_WORKER_LOST = True
```

**Running behind pgbouncer (transaction pooling).** The concurrency guard —
`select_for_update(nowait)` plus the partial unique constraint — works under
pgbouncer **transaction** pooling. Transaction mode does not support a few
PostgreSQL session features, so set these two options:

```python
DATABASES['default'].setdefault('OPTIONS', {})['prepare_threshold'] = None  # psycopg3: no server-side prepared statements
DATABASES['default']['DISABLE_SERVER_SIDE_CURSORS'] = True
```

Also do **not** force `sslmode=require` on the app→pgbouncer connection. That
hop is local and plaintext, and pgbouncer terminates TLS upstream. Without
`prepare_threshold=None`, the worker fails or hangs from time to time with
prepared-statement errors. This setup is validated end to end on Heroku behind an
in-dyno pgbouncer.

**Monitoring.** In Celery mode django-logic logs a failed attempt (`django-logic.transition` at ERROR) and records it on the row. It does **not** re-raise it as a Celery task exception, because that would flood your alerts and could re-deliver a row that is already resolved under `acks_late`. So watch the `TransitionMessage` table, not Celery task failures:

```sql
-- rows at the error limit (detect_stuck_transitions should be finalizing these)
SELECT count(*) FROM django_logic_background_transitionmessage
 WHERE is_completed = false AND errors_count >= 5;            -- = TRANSITION_MESSAGE_MAX_ERRORS

-- attempts running far longer than expected (watchdog candidates)
SELECT count(*) FROM django_logic_background_transitionmessage
 WHERE is_completed = false AND started_at < now() - interval '15 minutes';

-- rows superseded by an external state change (review these now and then:
-- each one is a manual fix or an external write that won over a waiting row)
SELECT count(*) FROM django_logic_background_transitionmessage
 WHERE last_error_message LIKE '[superseded]%';
```

Also alert when beat stops, because the safety net stops with it.

**Migrating an existing deployment.** Migration `0005` widens `instance_id` from integer to `varchar(255)` with `ALTER COLUMN ... TYPE`. Django emits the `USING ...::varchar` cast, so existing integer rows convert in place. On a very large `TransitionMessage` table this rewrites the column under a lock — run it in a maintenance window, or with your usual online-migration tooling. Migration `0006` (0.4.0) adds the `field_name` column. It also swaps the partial unique constraint from per-instance (`dl_bg_only_one_uncompleted_per_instance`) to per-process (`dl_bg_one_uncompleted_per_process`). That is a quick metadata and index change, safe to run in place.

## Testing Your Processes

FSM workflows are hard to test well, because states, conditions, permissions,
side-effects, background jobs, failures, retries and locking all interact.
`django_logic.testing` gives you a **scenario-based** test base class. A test
reads like the business process, and everything runs **inline, with no Celery
broker** — background transitions included.

Two principles keep these tests worth writing: **test your process, not the
machinery**, and **assert what the object became, not that a hook ran**. Full
rationale, and the scenario catalog, in
[docs/TESTING_GUIDE.md](docs/TESTING_GUIDE.md#journeys-not-mirrors).

```python
from django_logic.testing import ProcessScenario


class TestOrderFulfilment(ProcessScenario):
    """Order lifecycle: draft -> approved -> fulfilling -> fulfilled."""
    process_class = OrderProcess
    model = Order
    state_field = 'status'      # default: 'status'
    process_name = 'process'    # default: 'process'

    def test_happy_path(self):
        order = self.create_instance(status='approved')
        self.assert_available(order, ['fulfil', 'cancel'])

        self.background_transition(order, 'fulfil')      # enqueue + execute, no Celery
        self.assert_state(order, 'fulfilled')
        self.assert_side_effects_ran(['reserve_stock', 'call_courier'])
        self.assert_callbacks_ran(['send_confirmation_email'])

    def test_courier_failure_then_retry(self):
        order = self.create_instance(status='approved')

        # Make ONE named side-effect raise — the real failure path runs.
        self.background_transition(
            order, 'fulfil',
            fail_side_effect='call_courier',
            fail_with=ConnectionError('Aramex timeout'))

        self.assert_state(order, 'fulfilling')           # still in progress
        self.assert_error_recorded(order, 'Aramex timeout')
        self.assert_error_count(order, 1)
        self.assert_side_effects_not_ran(['call_courier'])

        self.retry_transition(order)                      # what the starter would do
        self.assert_state(order, 'fulfilled')

    def test_only_staff_can_approve(self):
        # self.staff / self.customer are your own setUp fixtures —
        # ProcessScenario does not create users.
        order = self.create_instance(status='draft')
        self.assert_available(order, ['approve'], user=self.staff)
        self.assert_not_available(order, ['approve'], user=self.customer)

    def test_approve_produces_the_right_outcome(self):
        # Assert what the object BECAME, not just that a hook ran.
        order = self.create_instance(status='draft')
        before = self.capture(order, ['status', 'approved_at'])
        self.transition(order, 'approve', user=self.staff)
        self.assert_side_effects_ran(['validate_order'])        # wiring
        self.assert_changed(order, before, {                    # outcome
            'status': ('draft', 'approved'),
            'approved_at': (None, order.approved_at),
        })

    def test_fulfil_failure_reaches_the_caller(self):
        order = self.create_instance(status='approved')
        # A failing side-effect runs the failure path AND re-raises — pin both.
        self.background_transition(order, 'fulfil',
                                   fail_side_effect='call_courier',
                                   fail_with=ConnectionError('down'),
                                   expect_raises=ConnectionError)
        self.assert_state(order, 'fulfilling')                  # still in progress
        self.assert_raised(ConnectionError, match='down')
```

**Driving the process**

| Method | What it does |
|--------|--------------|
| `create_instance(**fields)` | Create a model instance (state via the `state_field` kwarg) |
| `transition(obj, action, **kwargs)` | Run a synchronous transition |
| `background_transition(obj, action, **kwargs)` | Run a `BackgroundTransition`/`BackgroundAction` enqueue **and** execute inline |
| `retry_transition(obj)` | Re-run the instance's uncompleted transition — does what the periodic starter does |

Add `fail_side_effect='name'` and `fail_with=SomeError(...)` to `transition`, `background_transition` or `retry_transition` to make one named side-effect raise. django-logic wraps only that side-effect, so every other one runs for real and you exercise the true failure path. Add `expect_raises=SomeError` to assert the failure **reached the caller**, which is the contract for `side_effects`. Add `expect_raises=False` to assert django-logic **swallowed** it, which is the contract for `callbacks` and `next_transition`. Leave it out to absorb the injected exception and assert on the recorded error instead.

**Assertions**

- *State & availability:* `assert_state` · `assert_state_trace` · `assert_available` / `assert_not_available` (optional `user=`).
- *Domain outcome* — what the object *became*: `capture` → `assert_changed` / `assert_unchanged` · `assert_related_count`.
- *Wiring* — that a hook ran (pair with an outcome assertion): `assert_side_effects_ran` / `assert_side_effects_not_ran` · `assert_callbacks_ran` · `assert_failure_callbacks_ran`.
- *Caller boundary & durable row:* `assert_raised` / `assert_not_raised` · `assert_error_recorded` · `assert_error_count` · `assert_transition_owner`.
- *The whole journey:* `assert_journey([JourneyStep(...)])`.

django-logic **tracks** side-effects and callbacks by function `__name__`; it does not mock them. The real code runs and the framework records what ran. `assert_side_effects_ran` and `assert_callbacks_ran` check the wiring: a hook ran, not that it did the right thing. Pair them with `assert_changed`, `assert_related_count` or `assert_state`.

**Snapshot and replay.** `snapshot(obj)` captures an instance's fields, state and
`TransitionMessage` as JSON, from a shell, an admin action, Sentry or a log.
`self.from_snapshot('fixtures/bug_12345.json')` rebuilds it in a test. That turns
a production bug into a regression test. See the snapshot scenario in
[docs/TESTING_GUIDE.md](docs/TESTING_GUIDE.md).

**Failure output an agent can read.** When an assertion fails, the error carries a numbered timeline of every step and the relevant `TransitionMessage`. A person or an agent can then see where the process went wrong without reading a stack trace.

`ProcessScenario` extends `TransactionTestCase`, so it works with the durable `TransitionMessage` row and the atomic blocks around it.

## Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

### Development Setup

#### Option A: Local

1. Clone the repository
2. Create a virtual environment: `python -m venv venv`
3. Install dependencies: `pip install -e .`
4. Run tests: `python tests/manage.py test`

#### Option B: Docker + Make

The project includes a `Dockerfile` and a `makefile` so you can develop without installing anything locally.

```bash
make build          # build the Docker image
make test           # run the full test suite
make test-one t=tests.test_transition  # run a specific test module
make coverage       # run tests with coverage report
make sh             # open a Django shell inside the container
```

Please make sure to:
- Add tests for new features
- Update documentation
- Follow PEP 8 style guidelines
- Add type hints where applicable

## License
[MIT](https://choosealicense.com/licenses/mit/)

## Project status
Under active development. See [GitHub Issues](https://github.com/Borderless360/django-logic/issues) for planned features and known issues.

## Support
- 📖 [Documentation](https://github.com/Borderless360/django-logic/wiki)
- 🐛 [Issue Tracker](https://github.com/Borderless360/django-logic/issues)
- 💬 [Discussions](https://github.com/Borderless360/django-logic/discussions)
