menu

Nodes

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

Bases: viewflow.nodes.func.StartFunction

StartNode that can be executed within you code.

Example:

class MyFlow(Flow):
    start = flow.StartFunction(this.create_request)

    def create_request(self, activation, **kwargs):
        activation.prepare()
        activation.done()

MyFlow.create_request.run(**kwargs)

Note

Any kwarg you pass of the run call will be passed to the function.

Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activate_next_view_class

alias of ActivateNextTaskView

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

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.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.func.Function

Node that can be executed within you code.

Example:

class MyFlow(Flow):
    my_task = flow.Function(this.perform_my_task)

    @method_decorator(flow.flow_func(task_loader=lambda flow_task, **kwargs: ... ))
    def perform_my_task(self, activation, **kwargs):
        activation.prepare()
        activation.done()

MyFlow.my_task.run(**kwargs)

Note

Any kwarg you pass of the run call will be passed to the function.

Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activate_next_view_class

alias of ActivateNextTaskView

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Resolve internal this-referencies.

run(*args, **kwargs)

Execute the function task.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.handler.Handler

Node that can be executed automatically after task was created.

In difference to Function a Handler is not explicitly called in code, but executes automatically.

Example:

class MyFlow(Flow):
    my_task = (
        flow.Handler(this.handler_proc)
        .Next(this.End)
    )

    def my_handler(self. activation):
        # Your custom code
        pass

Note

You don’t need to call prepare() or done() on the activation in you handler callback.

Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activate_next_view_class

alias of ActivateNextTaskView

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Resolve internal this-referencies.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.ifgate.If

Activates one of paths based on condition.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

class viewflow.flow.Switch(**kwargs)

Bases: viewflow.nodes.switch.Switch

Activates first path with matched condition.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.join.Join

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

Join should be connected to one split task only.

Example:

join_on_warehouse = flow.Join().Next(this.next_task)
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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

class viewflow.flow.Split(**kwargs)

Bases: viewflow.nodes.split.Split

Activates outgoing path in-parallel depends on per-path condition.

Example:

split_on_decision = (
    flow.Split()
    .Next(check_post, cond=lambda p: p,is_check_post_required)
    .Next(this.perform_task_always)
)
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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.job.AbstractJob

Base task for background jobs.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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

Handle for url_type=’undo’.

job

Callable that should start the job in background.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.signal.StartSignal

StartNode that connects to a django signal.

Example:

def my_start_receiver(activation, **signal_kwargs):
    activation.prepare()
    # You custom code
    activation.done()

class MyFlow(Flow):
    start = flow.StartSignal(post_save, my_start_receiver, sender=MyModelCls)

Note

The first argument of your receiver will be the activation.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.signal.Signal

Node that connects to a django signal.

Example:

create_model = flow.Signal(post_create, my_receiver, sender=MyModelCls)

Note

Other than the StartSignal you will need to provide activation for your receiver yourself. This can be done using the flow_signal() decorator.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

undo_view_class

alias of UndoTaskView

urls()

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

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

Bases: viewflow.nodes.view.Start

Start process event.

Example:

start = (
    flow.Start(StartView, fields=["some_process_field"])
    .Available(lambda user: user.is_super_user)
    .Next(this.next_task)
)

In case of function based view:

start = flow.Start(start_process)

@flow_start_view()
def start_process(request, activation):
     if not activation.has_perm(request.user):
         raise PermissionDenied

     activation.prepare(request.POST or None)
     form = SomeForm(request.POST or None)

     if form.is_valid():
          form.save()
          activation.done()
          return redirect('/')
     return render(request, {'activation': activation, 'form': form})

Ensure to include {{ activation.management_form }} inside template, to proper track when task was started and other task performance statistics:

<form method="POST">
     {{ form }}
     {{ activation.management_form }}
     <button type="submit"/>
</form>
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.

activate_next_view_class

alias of ActivateNextTaskView

activation_class

alias of ManagedStartViewActivation

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

start_view_class

alias of CreateProcessView

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

Start view url.

view

View to perform flow start.

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

Bases: viewflow.nodes.view.View

View task.

Example:

task = (
    flow.View(some_view)
        .Permission('my_app.can_do_task')
        .Next(this.next_task)
)

In case of function based view:

task = flow.Task(task)

@flow_start_view()
def task(request, activation):
     if not activation.flow_task.has_perm(request.user):
         raise PermissionDenied

     activation.prepare(request.POST or None)
     form = SomeForm(request.POST or None)

     if form.is_valid():
          form.save()
          activation.done()
          return redirect('/')
     return render(request, {'activation': activation, 'form': form})

Ensure to include {{ activation.management_form }} inside template, to proper track when task was started and other task performance statistics:

<form method="POST">
     {{ form }}
     {{ activation.management_form }}
     <button type="submit"/>
</form>
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.

activate_next_view_class

alias of ActivateNextTaskView

activation_class

alias of ManagedViewActivation

assign_view

View to assign task to the user.

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

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

unassign_view_class

alias of UnassignTaskView

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

Add /assign/ and /unassign/ task urls.

view

View to perform user task.

class viewflow.flow.End(**kwargs)

Bases: viewflow.nodes.end.End

End process event.

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.

cancel_view_class

alias of CancelTaskView

detail_view

View for a task detail.

detail_view_class

alias of DetailTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

class viewflow.flow.Obsolete(*args, **kwargs)

Bases: viewflow.nodes.obsolete.Obsolete

Obsolete node removed from flow code definition, but loaded from db reference

This task can only be cancelled, without any additional actions

To enable obsolete nodes on flow, just include obsolete node in flow definition:

class MyFlow(Flow):
    ...
    obsolete = flow.Obsolete()

PRO Version only

activate(prev_activation, token)

Create task activation.

cancel_view_class

alias of CancelTaskView

create_node(name)

Create real node instance for missing entry

detail_view_class

alias of DetailTaskView

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

class viewflow.flow.StartSubprocess(func=None, **kwargs)

Bases: viewflow.nodes.subprocess.StartSubprocess

Next(node)

Next node to activate.

activate(prev_activation, token)

Create task activation.

activate_next_view_class

alias of ActivateNextTaskView

can_view(user, task)

Check if user has a view task detail permission.

cancel_view

View for the admin to cancel a task.

cancel_view_class

alias of CancelTaskView

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.

perform_view_class

alias of PerformTaskView

ready()

Resolve internal this-referencies.

run(*args, **kwargs)

Execute the function task.

undo_view

View for the admin to undo a task.

undo_view_class

alias of UndoTaskView

urls()

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

class viewflow.flow.Subprocess(subflow_task, **kwargs)

Bases: viewflow.nodes.subprocess.Subprocess

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.

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.flow.NSubprocess(subflow_task, subitem_source, **kwargs)

Bases: viewflow.nodes.subprocess.NSubprocess

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.

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.