menu

Nodes

API

class viewflow.rest.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.

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

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

def my_handler(activation):
    # Your custom code
    pass

class MyFlow(Flow):
    previous_task = flow.View(ProcessView)                 .Next(this.my_task)

    my_task = flow.Function(my_handler)                 .Next(this.End)

    end = flow.End()

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.

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

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

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

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.rest.flow.AbstractJob(job, **kwargs)

Bases: viewflow.nodes.job.AbstractJob

Base task fro 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.

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

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

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.rest.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) \
    .Activate(this.first_start)

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.

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

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.

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.

class viewflow.rest.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.

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

create_node(name)

Create real node instance for missing entry

ready()

Called when flow class setup finished.

Subclasses could perform additional initialisation here.

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

Bases: viewflow.nodes.subprocess.StartSubprocess

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=’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.rest.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.rest.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.