Writing Your Flow¶

This guide covers the main concepts for building workflows. If you’ve finished the quick start, this page explains how to design your own flows.

Data Management¶

Workflows have two kinds of data:

Flow data - State of the workflow: current step, decisions made, temporary values
Business data - Your application’s core data: users, orders, products

Keep them separate. Business data lives in your Django models. Flow data lives in the Process model.

Flow Data¶

The Process model has a data JSONField. Use viewflow.jsonstore to expose fields from that JSON:

from viewflow import jsonstore
from viewflow.workflow.models import Process


class HelloWorldProcess(Process):
    # Stored as process.data['approved'], accessible as process.approved
    approved = jsonstore.BooleanField(default=False)

    class Meta:
        proxy = True

Business Data¶

Store business data in separate models. The Process model has an artifact generic foreign key for linking to your business objects:

class MyModel(models.Model):
    message = models.CharField(max_length=150)


 def task_view(request, **kwargs):
     form = MyModelForm(request.POST or None)

     if form.is_valid():
         object = form.save(commit=True)
         request.activation.process.artifact = object

     # additional code here..

Task Data¶

Each task also has a data JSONField for task-specific information. This keeps data scoped to individual tasks rather than the whole process.

Starting a Flow¶

Interactive Start¶

Use flow.Start with a view. Viewflow provides CreateProcessView and CreateArtifactView:

from viewflow import this
from viewflow.workflow import flow
from viewflow.workflow.flow import views

class MyFlow(flow.Flow):
    start_with_artifact = (
        flow.Start(views.CreateArtifactView.as_view(model=MyModel, fields=['message']))
        .Annotation(title=_("Fill a form to start flow"))
        .Permission("myapp.can_start_request")
        .Next(this.next_task)
    )

Programmatic Start¶

Use flow.StartHandle to start flows from code:

from viewflow import this
from viewflow.workflow import flow

class MyFlow(flow.Flow):
    start_noninteractive = flow.StartHandle(this.start_process).Next(this.approve)

    def start_process(self, activation, message=''):
        object = MyModel.objects.create(message=message)
        activation.process.artifact = object
        return activation.process

# Start from anywhere in your code
process = MyFlow.start_noninteractive.run(message="Hello World")

User Tasks¶

User tasks need human input. Use flow.View with built-in views like UpdateProcessView or UpdateArtifactView:

class MyFlow(flow.Flow):
    ...

    approve = (
        flow.View(views.UpdateProcessView.as_view(fields=["approved"]))
        .Permission(auto_create=True)
        .Next(this.check_approve)
    )

Use flow.ManualTask() for work performed outside any system – the task appears in the task list and the user marks it done with a plain confirmation form:

greet_client = flow.ManualTask().Next(this.end)

Non-Interactive Tasks¶

flow.Function¶

Runs a function immediately when the previous task finishes. Runs in the same database transaction:

from viewflow import this
from viewflow.workflow import flow

class MyFlow(flow.Flow):
    process_class = MyProcess

    start = (
        flow.Start(...)
        .Next(this.process_data)
    )

    process_data = (
        flow.Function(this.process_data_function)
        .Next(this.end)
    )

    def process_data_function(self, activation):
        activation.process.sample_text = activation.process.sample_text.upper()
        activation.done()

    end = flow.End()

flow.SendHandle and flow.BusinessRule work the same way but export as dedicated BPMN task types – an outbound message hook (send task) and a rule evaluation (business rule task):

send = flow.SendHandle(this.notify_customer).Next(this.discount)
discount = flow.BusinessRule(this.calc_discount).Next(this.end)

flow.Handle¶

Waits for external code to call it. Use this for webhooks or external events:

from viewflow import this
from viewflow.workflow import flow

class MyFlow(flow.Flow):
    process_class = MyProcess

    start = (
        flow.Start(...)
        .Next(this.wait_for_external_event)
    )

    wait_for_external_event = (
        flow.Handle(this.handle_external_event)
        .Next(this.end)
    )

    def handle_external_event(self, activation, data):
        activation.process.sample_text = data['new_text']
        activation.done()

    end = flow.End()

# Call from external code
process = MyFlow.wait_for_external_event.run(
    process=my_process_instance,
    data={'new_text': 'Updated Text'}
)

celery.Job¶

Runs a task in Celery for long-running work:

# tasks.py
from celery import shared_task
from viewflow.flow import flow_job


@shared_task
def send_hello_world_request(activation_ref):
    with Job.activate(activation_ref) as activation:
        result = 'Background Processing Done'

        # No locks during long jobs - save carefully
        activation.process.sample_text = result
        activation.process.save(updated_fields=['sample_text'])

See also

Timer

flow.Timer¶

Waits until a moment stored in the database, so pending timers survive a message broker restart or flush:

class MyFlow(flow.Flow):
    ...
    wait = flow.Timer(timedelta(days=1)).Next(this.escalate)
    ...

The delay accepts a timedelta, an absolute datetime, or a callable activation -> timedelta | datetime. Due timers are fired by a periodic dispatcher – run the workflow_timers management command from cron, or schedule the viewflow.workflow.tasks.workflow_fire_timers task with celery beat:

app.conf.beat_schedule = {
    "viewflow-timers": {
        "task": "viewflow.workflow.tasks.workflow_fire_timers",
        "schedule": 60.0,
    },
}

flow.StartTimer¶

Starts a new process on a schedule, fired by the same dispatcher:

class ReportFlow(flow.Flow):
    start = flow.StartTimer(interval=timedelta(days=1)).Next(this.report)
    ...

For cron-style schedules, point celery beat or OS cron directly at MyFlow.start.run().

Intermediate Events¶

Intermediate events sit between tasks. Catch events wait; throw events fire and continue.

flow.MessageCatch waits for external code, like flow.Handle but exported as a message event; flow.MessageThrow runs an outbound hook:

wait_payment = flow.MessageCatch(this.on_payment).Next(this.notify)
notify = flow.MessageThrow(this.send_receipt).Next(this.end)

flow.SignalThrow broadcasts a named signal to every armed flow.SignalCatch, across all processes and flows – one throw releases them all:

# in one flow
shipped = flow.SignalThrow("order-shipped").Next(this.end)

# in another
wait = flow.SignalCatch("order-shipped").Next(this.invoice)

flow.ConditionalCatch waits until a condition over the process data becomes true. It is evaluated by the workflow_timers dispatcher, the same sweep that fires flow.Timer:

wait = flow.ConditionalCatch(
    lambda activation: activation.process.approved
).Next(this.proceed)

flow.EscalationThrow raised inside a subprocess notifies the parent’s non-interrupting .OnEscalation boundary without stopping either process – see Boundary Events below.

Boundary Events¶

A boundary event attaches to a task and fires while it is still active. Chain .OnTimeout / .OnError / .OnEscalation onto the host task, before .Next(). Each creates an auto-named boundary node (<task>__timeout, <task>__error, <task>__escalation):

class MyFlow(flow.Flow):
    approve = (
        flow.View(...)
        .OnTimeout(timedelta(days=3), this.escalate)
        .Next(this.end)
    )
    ...

.OnTimeout(delay, then) fires when the delay elapses first – deadlines and escalations. .OnError(then, code=...) fires when the host task fails in a background context (job, function, timer) and routes to a recovery path:

deploy = flow.Function(this.run_deploy).OnError(this.rollback).Next(this.end)

By default a boundary event interrupts (cancels) its host task; pass interrupting=False to start a parallel path and leave the host running. Pass title= to label the boundary in the chart. Boundary events are canceled automatically when the host task completes. Timer boundaries are fired by the same workflow_timers dispatcher as flow.Timer.

.OnEscalation(then, code=...) catches a flow.EscalationThrow raised inside a subprocess. It never interrupts – the subprocess and the child both keep running:

sub = (
    flow.Subprocess(OrderFlow.start)
    .OnEscalation(this.notify_manager, code="over-budget")
    .Next(this.end)
)

Gates¶

BPMN separates tasks (do something) from gates (decide what’s next). This makes workflows easier to understand and modify.

flow.If¶

Takes one of two paths based on a condition:

from viewflow import this
from viewflow.workflow import flow

class MyFlow(flow.Flow):
    ...

    check_condition = (
        flow.If(lambda activation: activation.process.approved)
        .Then(this.approved_task)
        .Else(this.rejected_task)
    )

    ...

See also

Switch for more than two branches

flow.Split and flow.Join¶

Split creates parallel branches. Join waits for all branches to finish:

from viewflow import this
from viewflow.workflow import flow

class MyFlow(flow.Flow):
    process_class = MyProcess

    start = (
        flow.Start(...)
        .Next(this.parallel_tasks)
    )

    parallel_tasks = (
        flow.Split()
        .Next(this.task1).Next(this.task2)
    )

    task1 = flow.View(...)
        .Next(this.join)

    task2 = flow.View(...)
        .Next(this.join)

    join = flow.Join()
        .Next(this.next_step)

    next_step = flow.End()

See also

SplitFirst

Subprocess¶

PRO-only

Use flow.Subprocess and flow.NSubprocess to break large flows into smaller, reusable pieces. flow.NSubprocess(..., sequential=True) runs one child process at a time instead of all at once.

See also

Subprocess

See also

NSubprocess

End¶

Use flow.End to finish a process. You can have multiple end nodes for different outcomes:

class MyFlow(flow.Flow):
    # previous flow steps...

    check_approve = (
        flow.If(act.process.approved)
        .Then(this.approved)
        .Else(this.rejected)
    )

    approved = flow.End()
    rejected = flow.End()

flow.TerminateEnd() cancels all other active tasks and finishes the process immediately, without waiting for parallel branches.

flow.ErrorEnd(code) does the same and records the process as failed. Inside a subprocess, the parent task is marked ERROR so the parent can catch it with .OnError(this.recover, code=...) on the subprocess.

Compensation¶

Register an undo handler on a task and run all handlers in reverse completion order with flow.CompensateThrow:

book_hotel = (
    flow.Function(this.do_book)
    .CompensateWith(this.cancel_hotel)
    .Next(this.pay)
)
cancel_hotel = flow.Function(this.do_cancel)
...
compensate = flow.CompensateThrow().Next(this.fail_end)
fail_end = flow.ErrorEnd("payment-failed")

Each completed task is compensated at most once. A handler is a flow.Function with no incoming connections; it receives the compensated task as activation.task.previous.

Diagram export¶

Every flow is exportable as a standard BPMN 2.0 file that opens in bpmn.io or Camunda Modeler. See BPMN Export.

Table Of Contents

Writing Your Flow¶

Data Management¶

Flow Data¶

Business Data¶

Task Data¶

Starting a Flow¶

Interactive Start¶

Programmatic Start¶

User Tasks¶

Non-Interactive Tasks¶

flow.Function¶

flow.Handle¶

celery.Job¶

flow.Timer¶

flow.StartTimer¶

Intermediate Events¶

Boundary Events¶

Gates¶

flow.If¶

flow.Split and flow.Join¶

Subprocess¶

End¶

Compensation¶

Diagram export¶