viewflow.workflow.flow.Start(view, undo_func=None, **kwargs)¶The Start node in a flow.
This node is used as the initial step in a flow by a user.
class MyFlow(flow.Flow):
start = (
flow.Start(views.CreateProcessView.as_view(fields=["text"]))
.Annotation(title=_("New message"))
.Permission(auto_create=True)
.Next(this.approve)
)
...
Next(node)¶Next node to activate.
Permission(permission=None, auto_create=False, obj=None, help_text=None)¶Make task available for users with specific permission.
Accepts permissions name or callable :: Callable[Activation] -> string:
.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.department.pk))
Task specific permission could be auto created during migration:
# Creates `process_class.can_do_task_process_class` permission
do_task = View().Permission(auto_create=True)
# You can specify permission codename and description right here
# The following creates `process_class.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)
can_execute(user, task=None)¶Check whether the user is authorized to start a flow.
This method checks whether the user is authorized to start a flow based on the owner and owner permission attributes.
can_view(user, task)¶Check if user has a view task detail permission.
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.StartHandle(func=None, undo_func=None)¶The Start handle node in a flow.
This node is used as the initial step in a flow from code
class MyFlow(flow.Flow):
start = flow.StartHandle(this.on_start_process).Next(this.approve)
def start_process(self, activation, sample=False):
activation.process.sample = sample
return activation.process
...
process = MyFlow.start.run(sample=True)
Next(node)¶Next node to activate.
can_view(user, task)¶Check if user has a view task detail permission.
detail_view_class¶alias of DetailTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.View(view, undo_func=None, **kwargs)¶Represents a user-interaction node within a flow
class MyFlow(flow.Flow):
...
approve = (
flow.View(views.UpdateProcessView.as_view(fields=["approved"]))
.Annotation(
title=_("Approve"),
description=_("Supervisor approvement"),
summary_template=_("Message review required"),
result_template=_(
"Message was {{ process.approved|yesno:'Approved,Rejected' }}"
),
)
.Permission(auto_create=True)
.Next(this.check_approve)
)
...
Assign(owner=None, **owner_kwargs)¶Assign task to the User immediately on activation.
Accepts user lookup kwargs or callable :: Process -> User:
.Assign(username='employee')
.Assign(lambda activation: activation.process.created_by)
Next(node)¶Next node to activate.
Permission(permission=None, auto_create=False, obj=None, help_text=None)¶Make task available for users with specific permission.
Accepts permissions name or callable :: Callable[Activation] -> string:
.Permission('my_app.can_approve')
.Permission(lambda process: 'my_app.department_manager_{}'.format(process.department.pk))
Task specific permission could be auto created during migration:
# Creates `process_class.can_do_task_process_class` permission
do_task = View().Permission(auto_create=True)
# You can specify permission codename and description right here
# The following creates `process_class.can_execute_task` permission
do_task = View().Permission('can_execute_task', help_text='Custom text', auto_create=True)
assign_view_class¶alias of AssignTaskView
calc_owner(activation)¶Determine a user to auto-assign the task.
calc_owner_permission(activation)¶Determine required permission to assign and execute this task.
calc_owner_permission_obj(activation)¶Determine required permission to assign and execute this task.
can_assign(user, task)¶Check if user can assign the task.
can_execute(user, task)¶Check user permission to execute the task.
can_unassign(user, task)¶Check if user can unassign the task.
can_view(user, task)¶Check if user has a view task detail permission.
cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of UserIndexTaskView
onCreate(ref)¶Call a function when task created:
class MyFlow(Flow):
approve = flow.View(...).OnCreate(this.on_approve_created)
def on_approve_created(self, activation):
if activation.task.owner:
send_mail(
'View task assigned to you','Here is the message.',
'from@example.com', [activation.task.owner.email]
)
revive_view_class¶alias of ReviveTaskView
unassign_view_class¶alias of UnassignTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.Function(func, **kwargs)¶detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.Handle(func=None, undo_func=None, **kwargs)¶Represents a task executed from the other parts of code
Usage: To define a handle in a flow and run it:
class MyFlow(flow.Flow):
...
my_handle = flow.Handle().Next(this.join_gate)
task = task=process.task_set.get(flow_task=MyFlow.my_handle, status=STATUS.NEW),
MyFlow.my_handle.run(task)
cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
undo_view_class¶alias of UndoTaskView
viewflow.contrib.celery.Job(celery_task, undo_func=None, *args, **kwargs)¶Run celery a task in background
Example.
tasks.py:
from celery import shared_task
from viewflow.flow import flow_job
@shared_task
def sample_task(activation):
...
flows.py:
from viewflow.contrib import celery
class MyFlow(Flow):
...
task = celery.Job(tasks.sample_task)
...._
Delay(delay)¶Async task execution delay
Eta(eta_callable)¶Expects callable that would get the task and return datetime for task execution
activation_class¶alias of JobActivation
cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of CeleryDetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.If(cond, **kwargs)¶The If-gate
class MyFlow(flow.Flow):
...
check_approve = (
flow.If(lambda activation: activation.process.approved)
.Annotation(title=_("Approvement check"))
.Then(this.send)
.Else(this.end)
)
. ...
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.Switch(**kwargs)¶cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.flow.Split(**kwargs)¶detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.Node(activation_class=None, **kwargs)¶Base class for a flow task definition.
Annotation(title=None, description=None, summary_template=None, result_template=None)¶Sets annotation for the node.
The node instance with the updated annotation values
get_available_actions(activation, user)¶Returns a list of available actions for the given user on the current node.
A list of available actions as a tuple of (name, url).
list
reverse(viewname, args=None, kwargs=None, current_app=None)¶Get view url.