menu

Nodes

Base abstract nodes declaration. Can be used as base classes for own nodes implementation.

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

Base class for flow task.

Parameters:
  • task_type – Human readable task type
  • activation_class – Activation implementation specific for this node
activate(prev_activation, token)

Create task activation.

get_task_url(task, url_type, **kwargs)

Return url for the task.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

urls()

List of urls for flow node views.

class viewflow.nodes.AbstractJob(job, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.DetailViewMixin, viewflow.Task

Base class for task that runs in background.

Example:

job = (
    flow.Job(task.job)
    .Next(this.end)
)
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’undo’.

job

Callable that should start the job in background.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/undo/ url.

class viewflow.nodes.End(**kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Event

End of the flow.

If no other parallel activities exists, finishes the whole process.

activate(prev_activation, token)

Create task activation.

activation_class

alias of EndActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Function(func, task_loader=None, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Event

Function task to be executed outside of the flow.

Example:

class MyFlow(Flow):
    ...

    shipment_received_handler = (
       flow.Function(
           this.on_shipment_receive,
           task_loader=this.get_shipment_handler_task)
       .Next(this.end)
   )

   ....

   @method_decorator(flow.flow_func)
   def on_shipment_receive(self, activation, shipment):
        activation.prepare()
        activation.done()

   def get_shipment_handler_task(self, flow_task, shipment):
        return Task.objects.get(process=shipment.process)

Somewhere in you code:

...
MyFlow.shipment_received_handler.run(shipment)
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of FuncActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Resolve internal this-referencies.

run(*args, **kwargs)

Execute the function task.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Handler(handler, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Event

Callback executed synchronously on a task activation.

Example:

class MyFlow(Flow):
    calc_total = (
        flow.Handler(this.calc_order_total)
    )

    ...

    def calc_order_total(self, activation):
        total = [
            sum(item.price)
            for item in activation.process.items.all()
        ]

        activation.process.total_amount = total
        activation.process.save()
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of HandlerActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Resolve internal this-referencies.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.If(cond, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Gateway

If gateway, activate one of outgoing node.

Example:

class MyFlow(Flow):
    check_approve = (
        flow.If(lambda activation: activation.process.is_approved)
        .Then(this.send_message)
        .Else(this.end_rejected)
    )
Else(node)

Node activated if condition is False.

Then(node)

Node activated if condition is True.

activate(prev_activation, token)

Create task activation.

activation_class

alias of IfActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Join(wait_all=True, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Gateway

Wait for one or all incoming links and activates next path.

Without a Join, subsequent task would be activated as many times as it have parallel incoming links.

Example:

class MyFlow(Flow):
    prepare_item = (
        flow.Split()
        .Next(this.make_box)
        .Next(this.make_label)
    )

    make_box = (
        flow.View(ConfirmView, fields=['box_done'])
        .Next(this.item_prepared)
    )

    make_label = (
        flow.View(ConfirmView, fields=['label_done'])
        .Next(this.item_prepared)
    )

    item_prepared = flow.Join().Next(this.end)
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of JoinActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Signal(signal, receiver, sender=None, task_loader=None, allow_skip=False, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.Event

Execute a callback on a django signal receive.

Example:

class MyFlow(Flow):
    wait_for_receipt = (
        flow.Signal(
            post_create, this.receipt_created,
            sender=MyModelCls)
        .Next(this.approve)

    ...

    def receipt_created(self, activation, **signal_kwargs):
        activation.prepare()
        activation.process.receipt = signal_kwargs['instance']
        activation.done()
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of FuncActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

on_signal(sender, **signal_kwargs)

Signal handler.

ready()

Resolve internal this-referencies. and subscribe to the signal.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Split(**kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Gateway

Parallel split gateway.

Activates outgoing tasks to execute concurently. Assumes that all outgoing path converges at the same Join node.

If Split has no nodes to activate, FlowRuntimeError would be raised.

Example:

split_clerk_warehouse = (
    flow.Split()
    .Next(
        this.package_goods,
        cond=lambda a: a.process.need_packaging)
    .Always(this.shipment_type)
)
Always(node)

Activate that node unconditionally.

Shortcut for .Next without a condition

Next(node, cond=None)

Node to activate if condition is true.

Parameters:cond – Callable[activation] -> bool
activate(prev_activation, token)

Create task activation.

activation_class

alias of SplitActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Start(*args, **kwargs)

Bases: viewflow.mixins.PermissionMixin, viewflow.nodes.view.BaseStart

User task that starts flow from a django view.

Available(owner=None, **owner_kwargs)

Make process start action available for the User.

Accepts user lookup kwargs or callable predicate :: User -> bool:

.Available(username='employee')
.Available(lambda user: user.is_super_user)
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.

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

.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.depratment.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 `processcls_app.can_execure_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)
activate(prev_activation, token)

Create task activation.

activate_next_view

View for the admin to perform activate action.

activation_class

alias of StartActivation

can_execute(user, task=None)

Check user permission to start a flow.

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

“Handle url_Type=’execute’.

If url_type is ‘guess’ and task can be executed by user, the ‘execute’ url is returned.

ready()

Insert additional flow permissions to the meta of the process model.

Permissions itself are created as usual during django database migration process.

undo_view

View for the admin to undo a task.

urls()

Start view url.

view

View to perform flow start.

class viewflow.nodes.StartFunction(func=None, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Event

Start function task to state a flow from the code.

Example:

class MyFlow(Flow):
   start = (
       flow.StartFunction(this.start_flow)
       .Next(this.end)
   )

   ....

   @method_decorator(flow.flow_start_func)
   def on_shipment_receive(self, activation, shipment):
        activation.prepare()
        activation.process.shipment = shipment
        activation.done()
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of StartActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Resolve internal this-referencies.

run(*args, **kwargs)

Execute the function task.

start_func_default(*args, **kwargs)

Default implementation for start function.

Do nothing, just create a new process instance.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.StartSignal(signal, receiver, sender=None, **kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.NextNodeMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.Event

Start flow on a django signal receive.

Example:

class MyFlow(Flow):
    start = (
        flow.StartSignal(
            post_save, this.start_flow,
            sender=MyModelCls)
        .Next(this.approve)
    )

    ...

    @flow_start_signal
    def start_flow(self, activation, **signal_kwargs):
        activation.prepare()
        activation.done()
Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activation_class

alias of StartActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

on_signal(sender, **signal_kwargs)

Signal handler.

ready()

Resolve internal this-referencies. and subscribe to the signal.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Switch(**kwargs)

Bases: viewflow.mixins.TaskDescriptionMixin, viewflow.mixins.DetailViewMixin, viewflow.mixins.UndoViewMixin, viewflow.mixins.CancelViewMixin, viewflow.mixins.PerformViewMixin, viewflow.Gateway

Gateway that selects one of the outgoing node.

Activates first node with matched condition.

Example:

select_responsible_person = (
    flow.Switch()
    .Case(this.dean_approval, lambda act: a.process.need_dean)
    .Case(this.head_approval, lambda act: a.process.need_head)
    .Default(this.supervisor_approval)
)
Case(node, cond=None)

Node to activate if condition is True.

Parameters:cond – Calable[activation] -> bool
Default(node)

Last node to activate if no one other succeed.

activate(prev_activation, token)

Create task activation.

activation_class

alias of SwitchActivation

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

perform_view

View for the admin to re-execute a gate.

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.View(*args, **kwargs)

Bases: viewflow.mixins.PermissionMixin, viewflow.nodes.view.BaseView

User Task that can be executed in a django view.

Example:

class MyFlow(Flow):
    task = (
        flow.View(some_view)
        .Permission('my_app.can_do_task')
        .Next(this.next_task)
    )
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 process: 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.

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

.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.depratment.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 `processcls_app.can_execure_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)
activate(prev_activation, token)

Create task activation.

activate_next_view

View for the admin to perform activate action.

activation_class

alias of ViewActivation

assign_view

View to assign task to the user.

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 premition 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

View for the admin to cancel a task.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle assign, unassign and execute url_types.

If url_type is guess task check is it can be assigned, unassigned or executed. If true, the action would be returned as guess result url.

ready()

Insert additional flow permissions to the meta of the process model.

Permissions itself are created as usual during django database migration process.

unassign_view

View to unassign task from the user.

undo_view

View for the admin to undo a task.

urls()

Add /assign/ and /unassign/ task urls.

view

View to perform user task.

Base nodes

class viewflow.Event(activation_class=None, **kwargs)

Bases: viewflow.Node

Base class for event-based tasks.

class viewflow.Task(activation_class=None, **kwargs)

Bases: viewflow.Node

Base class for tasks.

class viewflow.Gateway(activation_class=None, **kwargs)

Bases: viewflow.Node

Base class for task gateways.

Node mixins

class viewflow.mixins.NextNodeMixin(*args, **kwargs)

Mixin for nodes that have only one outgoing path.

Next(node)

Next node to activate.

class viewflow.mixins.DetailViewMixin(*args, **kwargs)

Task details.

can_view(user, task)

Check if user has a view task detail permission.

detail_view

View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’detail’.

The end of url_type=`guess` chain. If all previous mixins does not return a value, the details page would be used.

urls()

Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.mixins.UndoViewMixin(*args, **kwargs)

Undo a completd task.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’undo’.

undo_view

View for the admin to undo a task.

urls()

Add /<process_pk>/<task_pk>/undo/ url.

class viewflow.mixins.CancelViewMixin(*args, **kwargs)

Cancel a task action.

cancel_view

View for the admin to cancel a task.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’cancel’.

urls()

Add /<process_pk>/<task_pk>/cancel/ url.

class viewflow.mixins.PerformViewMixin(*args, **kwargs)

Re-execute a gate manually.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle for url_type=’perform’.

perform_view

View for the admin to re-execute a gate.

urls()

Add /<process_pk>/<task_pk>/perform/ url.

class viewflow.mixins.ActivateNextMixin(*args, **kwargs)

Mixing allows administrator manually activate outgoing nodes.

activate_next_view

View for the admin to perform activate action.

get_task_url(task, url_type='guess', namespace='', **kwargs)

Handle url_type=’activate_next’.

urls()

Add /<process_pk>/<task_pk>/activate_next/ url.

class viewflow.mixins.PermissionMixin(*args, **kwargs)

Node mixin to restrict access using django permissions.

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

Make task available for users with specific permission.

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

.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.depratment.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 `processcls_app.can_execure_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)
ready()

Insert additional flow permissions to the meta of the process model.

Permissions itself are created as usual during django database migration process.

class viewflow.mixins.TaskDescriptionMixin(view_or_class=None, task_title=None, task_description=None, task_result_summary=None, **kwargs)

Task explanation for the end user.

Parameters:
  • task_title – The task brief
  • task_description – Details what should be done
  • task_result_summary – Details how task was executed.

task_result_summary could be a django template string. The template is rendered with following context variables:

['process', 'task', 'flow_class', 'flow_task']
class viewflow.mixins.TaskDescriptionViewMixin(view_or_class=None, **kwargs)

Extract task desctiption from the view docstring.

class viewflow.mixins.ViewArgsMixin(**kwargs)

Capture rest of kwargs as view kwargs.

Put this mixing always the last in inheritance order