Built-in Nodes

Start a Flow

class viewflow.workflow.flow.Start(view, undo_func=None, **kwargs)

The Start node in a flow.

This node is used as the initial step in a flow by a user.

Live Demo / Cookbook sample

class MyFlow(flow.Flow):
    start = (
        flow.Start(views.CreateProcessView.as_view(fields=["text"]))
        .Annotation(title=_("New message"))
        .Permission(auto_create=True)
        .Next(this.approve)
    )

    ...

Next(node)

Next node to activate.

Permission(permission=None, auto_create=False, obj=None, help_text=None)

Make task available for users with specific permission.

Accepts permissions name or callable :: Callable[Activation] -> string:

.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.department.pk))

Task specific permission could be auto created during migration:

# Creates `process_class.can_do_task_process_class` permission
do_task = View().Permission(auto_create=True)

# You can specify permission codename and description right here
# The following creates `process_class.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)

can_execute(user, task=None)

Check whether the user is authorized to start a flow.

This method checks whether the user is authorized to start a flow based on the owner and owner permission attributes.

can_view(user, task)

Check if user has a view task detail permission.

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

undo_view_class

alias of UndoTaskView

class viewflow.workflow.flow.StartHandle(func=None, undo_func=None)

The Start handle node in a flow.

This node is used as the initial step in a flow from code

class MyFlow(flow.Flow):
    start = flow.StartHandle(this.on_start_process).Next(this.approve)

    def start_process(self, activation, sample=False):
        activation.process.sample = sample
        return activation.process

    ...

process = MyFlow.start.run(sample=True)

Next(node)

Next node to activate.

can_view(user, task)

Check if user has a view task detail permission.

detail_view_class

alias of DetailTaskView

undo_view_class

alias of UndoTaskView

User Task

class viewflow.workflow.flow.View(view, undo_func=None, **kwargs)

Represents a user-interaction node within a flow

class MyFlow(flow.Flow):
    ...

    approve = (
        flow.View(views.UpdateProcessView.as_view(fields=["approved"]))
        .Annotation(
            title=_("Approve"),
            description=_("Supervisor approvement"),
            summary_template=_("Message review required"),
            result_template=_(
                "Message was {{ process.approved|yesno:'Approved,Rejected' }}"
            ),
        )
        .Permission(auto_create=True)
        .Next(this.check_approve)
    )

    ...

Assign(owner=None, **owner_kwargs)

Assign task to the User immediately on activation.

Accepts user lookup kwargs or callable :: Process -> User:

.Assign(username='employee')
.Assign(lambda activation: activation.process.created_by)

Next(node)

Next node to activate.

Permission(permission=None, auto_create=False, obj=None, help_text=None)

Make task available for users with specific permission.

Accepts permissions name or callable :: Callable[Activation] -> string:

.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.department.pk))

Task specific permission could be auto created during migration:

# Creates `process_class.can_do_task_process_class` permission
do_task = View().Permission(auto_create=True)

# You can specify permission codename and description right here
# The following creates `process_class.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)

assign_view_class

alias of AssignTaskView

calc_owner(activation)

Determine a user to auto-assign the task.

calc_owner_permission(activation)

Determine required permission to assign and execute this task.

calc_owner_permission_obj(activation)

Determine required permission to assign and execute this task.

can_assign(user, task)

Check if user can assign the task.

can_execute(user, task)

Check user permission to execute the task.

can_unassign(user, task)

Check if user can unassign the task.

can_view(user, task)

Check if user has a view task detail permission.

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of DetailTaskView

index_view_class

alias of UserIndexTaskView

onCreate(ref)

Call a function when task created:

class MyFlow(Flow):
    approve = flow.View(...).OnCreate(this.on_approve_created)

    def on_approve_created(self, activation):
        if activation.task.owner:
            send_mail(
                'View task assigned to you','Here is the message.',
                'from@example.com', [activation.task.owner.email]
            )

revive_view_class

alias of ReviveTaskView

unassign_view_class

alias of UnassignTaskView

undo_view_class

alias of UndoTaskView

Script Tasks

class viewflow.workflow.flow.Function(func, **kwargs)

Represents a callback function executed synchronously as part of a workflow node.

A Function node is used within a flow to execute a callable (e.g., a method) that operates on the process instance. The execution is synchronous, meaning the workflow will wait for the callable to complete before proceeding to the next node.

Usage:

In the following example, a Function node is used within a MyFlow class to execute a logging function immediately after a task been activated.

class MyFlow(flow.Flow):
    ...

    log_immediately = (
        flow.Function(this.log) .Next(this.end)
    )

    def log(self, activation):
        print(f"Process is in action {activation.process.pk}")

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

undo_view_class

alias of UndoTaskView

class viewflow.workflow.flow.Handle(func=None, undo_func=None, **kwargs)

Represents a task executed from the other parts of code

Usage: To define a handle in a flow and run it:

class MyFlow(flow.Flow):
    ...
    my_handle = flow.Handle().Next(this.join_gate)

task = task=process.task_set.get(flow_task=MyFlow.my_handle, status=STATUS.NEW),
MyFlow.my_handle.run(task)

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

Job

class viewflow.contrib.celery.Job(celery_task, undo_func=None, *args, **kwargs)

Run celery a task in background

Example.

tasks.py:

from celery import shared_task
from viewflow.flow import flow_job


@shared_task
def sample_task(activation):
    ...

flows.py:

from viewflow.contrib import celery

class MyFlow(Flow):
    ...
    task = celery.Job(tasks.sample_task)
    ...._

Delay(delay)

Async task execution delay

Eta(eta_callable)

Expects callable that would get the task and return datetime for task execution

activation_class

alias of JobActivation

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of CeleryDetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

Gates

class viewflow.workflow.flow.If(cond, **kwargs)

The If-gate

   class MyFlow(flow.Flow):
       ...

       check_approve = (
           flow.If(lambda activation: activation.process.approved)
           .Annotation(title=_("Approvement check"))
           .Then(this.send)
           .Else(this.end)
       )

.       ...

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

class viewflow.workflow.flow.Switch(**kwargs)

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

class viewflow.workflow.flow.Split(**kwargs)

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

class viewflow.workflow.flow.SplitFirst(**kwargs)

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

class viewflow.workflow.flow.Join(continue_on_condition=None, **kwargs)

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of DetailTaskView

index_view_class

alias of IndexTaskView

revive_view_class

alias of ReviveTaskView

undo_view_class

alias of UndoTaskView

Timer

class viewflow.contrib.celery.Timer(delay, task=None, *args, **kwargs)

Wait till specified time interval

flows.py:

from viewflow.contrib import celery

class MyFlow(Flow):
    ...
    wait = celery.Timer(timedelta(minutes=10)).Next(this.check)
    ...._

activation_class

alias of TimerActivation

cancel_view_class

alias of CancelTaskView

detail_view_class

alias of CeleryDetailTaskView

revive_view_class

alias of ReviveTaskView

Base class

class viewflow.workflow.Node(activation_class=None, **kwargs)

Base class for a flow task definition.

Parameters

• task_type – Human readable task type
• activation_class – Activation implementation specific for this node

Annotation(title=None, description=None, summary_template=None, result_template=None)

Sets annotation for the node.

Parameters

• title – The title for the task
• description – The description for the task
• summary_template – The template for the task summary
• result_template – The template for the task result

Returns

The node instance with the updated annotation values

get_available_actions(activation, user)

Returns a list of available actions for the given user on the current node.

Parameters

• activation (viewflow.models.Activation) – The current activation instance.
• user (django.contrib.auth.models.User) – The user to check actions for.

Returns

A list of available actions as a tuple of (name, url).

Return type

list

reverse(viewname, args=None, kwargs=None, current_app=None)

Get view url.