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 initialization 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.

property cancel_view¶: View for the admin to cancel a task.

property detail_view¶: View for a task detail.

get_task_url(task, url_type='guess', namespace='', **kwargs)¶: Handle for url_type=’undo’.

property job¶: Callable that should start the job in background.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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 viewflow.activation.EndActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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 viewflow.activation.FuncActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶: Resolve internal this-references.

run(*args, **kwargs)¶: Execute the function task.

property undo_view¶: View for the admin to undo a task.

urls()¶: Add /<process_pk>/<task_pk>/detail/ url.

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

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 viewflow.nodes.handler.HandlerActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶: Resolve internal this-references.

property undo_view¶: View for the admin to undo a task.

urls()¶: Add /<process_pk>/<task_pk>/detail/ url.

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

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 viewflow.nodes.ifgate.IfActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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)¶

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 viewflow.nodes.join.JoinActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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 viewflow.activation.FuncActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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-references. and subscribe to the signal.

property undo_view¶: View for the admin to undo a task.

urls()¶: Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Split(**kwargs)¶

Parallel split gateway.

Activates outgoing tasks to execute concurrently. 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 viewflow.nodes.split.SplitActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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.

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 `processcls_app.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)

activate(prev_activation, token)¶: Create task activation.

property activate_next_view¶: View for the admin to perform activate action.

activation_class¶: alias of viewflow.activation.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.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property undo_view¶: View for the admin to undo a task.

urls()¶: Start view url.

property view¶: View to perform flow start.

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

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 viewflow.activation.StartActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶: Resolve internal this-references.

run(*args, **kwargs)¶: Execute the function task.

start_func_default(activation)¶

Default implementation for start function.

Do nothing, just create a new process instance.

property 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 viewflow.activation.StartActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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-references. and subscribe to the signal.

property undo_view¶: View for the admin to undo a task.

urls()¶: Add /<process_pk>/<task_pk>/detail/ url.

class viewflow.nodes.Switch(**kwargs)¶

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 viewflow.nodes.switch.SwitchActivation

can_view(user, task)¶: Check if user has a view task detail permission.

property cancel_view¶: View for the admin to cancel a task.

property 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.

property perform_view¶: View for the admin to re-execute a gate.

ready()¶

Called when flow class setup finished.

Subclasses could perform additional initialization here.

property 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.

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 `processcls_app.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)

activate(prev_activation, token)¶: Create task activation.

property activate_next_view¶: View for the admin to perform activate action.

activation_class¶: alias of viewflow.activation.ViewActivation

property 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.

property cancel_view¶: View for the admin to cancel a task.

property 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.

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]
            )

ready()¶: Resolve internal this-references.

property unassign_view¶: View to unassign task from the user.

property undo_view¶: View for the admin to undo a task.

urls()¶: Add /assign/ and /unassign/ task urls.

property 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.

property 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 completed task.

get_task_url(task, url_type='guess', namespace='', **kwargs)¶: Handle for url_type=’undo’.

property 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.

property 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’.

property 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.

property 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.

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 `processcls_app.can_execute_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 description 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