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)
)
...
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, task_data: Callable[[Activation], Dict[str, Any]] | None = None, task_seed: Callable[[Activation], Any] | None = None)¶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, task_data: Callable[[Activation], Dict[str, Any]] | None = None, task_seed: Callable[[Activation], Any] | None = None)¶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.StartTimer(interval=None, **kwargs)¶Start a new process on a schedule, fired by the workflow_timers
dispatcher.
Example:
class ReportFlow(flow.Flow):
start = flow.StartTimer(interval=timedelta(days=1)).Next(this.report)
...
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, task_data: Callable[[Activation], Dict[str, Any]] | None = None, task_seed: Callable[[Activation], Any] | None = None)¶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.ManualTask(view=None, **kwargs)¶Task performed by a person outside of any system (BPMN manual task).
Shows up in the task list; the user picks it and marks it done – by default with a plain confirmation form, no fields.
Example:
greet = flow.ManualTask().Next(this.end)
viewflow.workflow.flow.Function(func, **kwargs)¶Represents a callback function executed synchronously as part of a workflow node.
A Function node is used within a flow to execute a callable (e.g., a method) that operates on the process instance. The execution is synchronous, meaning the workflow will wait for the callable to complete before proceeding to the next node.
Usage:
In the following example, a Function node is used within a MyFlow class to execute a logging function immediately after a task been activated.
class MyFlow(flow.Flow):
...
log_immediately = (
flow.Function(this.log) .Next(this.end)
)
def log(self, activation):
print(f"Process is in action {activation.process.pk}")
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.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.workflow.flow.SendHandle(func, **kwargs)¶Outbound message hook, executed synchronously (BPMN send task).
Example:
send = flow.SendHandle(this.notify_customer).Next(this.end)
viewflow.workflow.flow.BusinessRule(func, **kwargs)¶Business rule evaluation, executed synchronously (BPMN business rule task).
Example:
discount = flow.BusinessRule(this.calc_discount).Next(this.end)
viewflow.workflow.flow.Subprocess(start_subprocess_task: StartHandle, get_subprocess_kwargs: Callable[[Activation], Dict[str, Any]] | None = None, process_data: Callable[[Activation], Dict[str, Any]] | None = None, task_data: Callable[[Activation], Dict[str, Any]] | None = None, process_seed: Callable[[Activation], Any] | None = None, task_seed: Callable[[Activation], Any] | None = None, **kwargs)¶The Subprocess node in a flow (PRO-only)
This node is used to start a subprocess flow within a parent flow. The subprocess must be completed before the parent flow can proceed.
class ExampleSubFlow(flow.Flow):
start = flow.StartHandle(this.start_func).Next(this.task)
task = flow.Handle(this.task_func).Next(this.end)
end = flow.End()
def start_func(self, activation):
# get access to parent process and data
activation.process.parent_task.process.data
def task_func(self, activation):
pass
class MainFlowWithSubprocess(flow.Flow):
start = flow.StartHandle().Next(this.subprocess)
subprocess = flow.Subprocess(ExampleSubFlow.start).Next(this.end)
end = flow.End()
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.NSubprocess(start_subprocess_task, subitem_source, sequential=False, **kwargs)¶The NSubprocess node in a flow (PRO-only)
This node is used to start multiple instances of a subprocess flow within a parent flow. Each instance processes a different item, and all subprocesses must be completed before the parent flow can proceed.
class ExampleSubFlow(flow.Flow):
start = flow.StartHandle(this.start_func).Next(this.task) task =
flow.Handle(this.task_func).Next(this.end)
end = flow.End()
def start_func(self, activation, item=0):
# instantialed with one of 1, 2, 3, 4 as item
activation.process.data = item
def task_func(self, activation):
activation.process.data += 100
class MainFlowWithNSubprocess(flow.Flow):
start = flow.StartFunction().Next(this.nsubprocess) nsubprocess =
flow.NSubprocess(ExampleSubFlow.start, lambda p: [1, 2, 3, 4]).Next(this.end)
end = flow.End()
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_ref):
with celery.Job.activate(activation_ref) as activation:
print(activation.process.message)
...
flows.py:
from viewflow.contrib import celery
class MyFlow(Flow):
...
task = celery.Job(tasks.sample_task)
....
For more information, see the Celery
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)
)
. ...
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.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)
)
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)¶Represents a parallel split gateway in a workflow, allowing branching into multiple parallel paths.
Example:
flow.Split()
.Next(
this.approve,
case=act.process.approved,
data_source=lambda activation: [{"sample": "test task 1"}, {"sample": "test task 2"}],
)
.Always(this.required)
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.SplitFirst(**kwargs)¶Parallel split, as soon as the first task is completed, the remaining tasks are cancelled.
The SplitFirst class is useful in workflows where you want to initiate multiple parallel tasks but only require the first task to complete, cancelling the rest once the first task finishes.
Example:
class MyFlow(flow.Flow):
split_first = SplitFirst().Next(this.task_a).Next(this.task_b)
task_a = flow.View(views.UserView).Next(this.join)
task_b = celery.Timer(delay=timedelata(minutes=10)).Next(this.join)
join = flow.Join()
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.Join(continue_on_condition=None, cancel_active=True, **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.Subprocess(start_subprocess_task: StartHandle, get_subprocess_kwargs: Callable[[Activation], Dict[str, Any]] | None = None, process_data: Callable[[Activation], Dict[str, Any]] | None = None, task_data: Callable[[Activation], Dict[str, Any]] | None = None, process_seed: Callable[[Activation], Any] | None = None, task_seed: Callable[[Activation], Any] | None = None, **kwargs)¶The Subprocess node in a flow (PRO-only)
This node is used to start a subprocess flow within a parent flow. The subprocess must be completed before the parent flow can proceed.
class ExampleSubFlow(flow.Flow):
start = flow.StartHandle(this.start_func).Next(this.task)
task = flow.Handle(this.task_func).Next(this.end)
end = flow.End()
def start_func(self, activation):
# get access to parent process and data
activation.process.parent_task.process.data
def task_func(self, activation):
pass
class MainFlowWithSubprocess(flow.Flow):
start = flow.StartHandle().Next(this.subprocess)
subprocess = flow.Subprocess(ExampleSubFlow.start).Next(this.end)
end = flow.End()
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.NSubprocess(start_subprocess_task, subitem_source, sequential=False, **kwargs)¶The NSubprocess node in a flow (PRO-only)
This node is used to start multiple instances of a subprocess flow within a parent flow. Each instance processes a different item, and all subprocesses must be completed before the parent flow can proceed.
class ExampleSubFlow(flow.Flow):
start = flow.StartHandle(this.start_func).Next(this.task) task =
flow.Handle(this.task_func).Next(this.end)
end = flow.End()
def start_func(self, activation, item=0):
# instantialed with one of 1, 2, 3, 4 as item
activation.process.data = item
def task_func(self, activation):
activation.process.data += 100
class MainFlowWithNSubprocess(flow.Flow):
start = flow.StartFunction().Next(this.nsubprocess) nsubprocess =
flow.NSubprocess(ExampleSubFlow.start, lambda p: [1, 2, 3, 4]).Next(this.end)
end = flow.End()
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.Timer(delay, **kwargs)¶Database-backed wait until a specified time interval or moment.
Due timers are fired by the periodic workflow_timers dispatcher.
Example:
class MyFlow(flow.Flow):
...
wait = flow.Timer(timedelta(days=1)).Next(this.escalate)
...
cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
viewflow.contrib.celery.Timer(delay, task=None, *args, **kwargs)¶Wait till specified time interval
flows.py:
from viewflow.contrib import celery
class MyFlow(Flow):
...
wait = celery.Timer(timedelta(minutes=10)).Next(this.check)
...._
activation_class¶alias of TimerActivation
cancel_view_class¶alias of CancelTaskView
detail_view_class¶alias of CeleryDetailTaskView
index_view_class¶alias of IndexTaskView
revive_view_class¶alias of ReviveTaskView
viewflow.workflow.flow.MessageCatch(func=None, undo_func=None, **kwargs)¶Intermediate message catch event, completed from external code.
Example:
wait_payment = flow.MessageCatch(this.on_payment).Next(this.ship)
# from external code
MyFlow.wait_payment.run(task, amount=100)
viewflow.workflow.flow.MessageThrow(func, **kwargs)¶Intermediate message throw event – an outbound hook executed synchronously.
Example:
notify = flow.MessageThrow(this.send_receipt).Next(this.end)
viewflow.workflow.flow.SignalCatch(name, **kwargs)¶Intermediate signal catch event: waits until a matching signal is thrown by any process.
Example:
wait = flow.SignalCatch("order-shipped").Next(this.notify)
viewflow.workflow.flow.SignalThrow(name, **kwargs)¶Intermediate signal throw event: broadcasts a named signal to every armed
flow.SignalCatch, then continues.
Example:
fire = flow.SignalThrow("order-shipped").Next(this.end)
viewflow.workflow.flow.ConditionalCatch(condition, **kwargs)¶Intermediate conditional catch event: waits until a condition over the
process data holds, evaluated by the workflow_timers dispatcher.
Example:
wait = flow.ConditionalCatch(
lambda activation: activation.process.approved
).Next(this.proceed)
viewflow.workflow.flow.EscalationThrow(code=None, **kwargs)¶Escalation throw event raised inside a subprocess, caught by the parent’s
.OnEscalation boundary. Nothing is interrupted.
Example:
over_budget = flow.EscalationThrow("over-budget").Next(this.keep_going)
Boundary events are declared fluently on the host task with .OnTimeout,
.OnError and .OnEscalation (chained before .Next()). See
Node.
viewflow.workflow.flow.End(activation_class: type | None = None, **kwargs: Any)¶The End node in a flow.
This node serves as the terminal point of a flow
class MyFlow(flow.Flow):
...
approved = this.End()
rejected = 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.TerminateEnd(activation_class: type | None = None, **kwargs: Any)¶Terminate end event: cancels all other active tasks and finishes the process immediately.
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.ErrorEnd(code=None, **kwargs)¶Error end event: interrupts the process and records it as failed. In a
subprocess, marks the parent task ERROR so the parent’s
.OnError boundary can catch it.
Example:
fail_end = flow.ErrorEnd("payment-failed")
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.CompensateThrow(*args, **kwargs)¶Compensation throw event: runs the .CompensateWith handlers of
completed tasks in reverse completion order, then continues.
detail_view_class¶alias of DetailTaskView
index_view_class¶alias of IndexTaskView
undo_view_class¶alias of UndoTaskView
viewflow.workflow.Node(activation_class: type | None = None, **kwargs: Any)¶Base class for a flow task definition.
Annotation(title: str | None = None, description: str | None = None, summary_template: str | None = None, result_template: str | None = None) Node¶Sets annotation for the node.
The node instance with the updated annotation values
CompensateWith(node: Any) Node¶Register a compensation handler for this task.
The handler is a flow.Function node without incoming connections,
executed by flow.CompensateThrow when the flow compensates.
OnError(then: Any, code: Any = None, interrupting: bool = True, title: str | None = None) Node¶Attach an error boundary event to this task.
Fires the then path when the task fails in a background context
(job, function, timer). Pass code to catch only a matching
flow.ErrorEnd code. The boundary node is auto-named
<task>__error; pass title to set its chart label. Chain
before .Next().
OnEscalation(then: Any, code: Any = None, title: str | None = None) Node¶Attach a non-interrupting escalation boundary event to this task.
Fires the then path when a flow.EscalationThrow is raised
inside this subprocess, without interrupting it. The boundary node is
auto-named <task>__escalation; pass title to set its chart
label. Chain before .Next().
OnTimeout(delay: Any, then: Any, interrupting: bool = True, title: str | None = None) Node¶Attach a timer boundary event to this task.
Fires the then path when delay elapses before the task
completes. Interrupting by default (cancels the task);
interrupting=False starts a parallel path and leaves it running.
The boundary node is auto-named <task>__timeout; pass title
to set its chart label. Chain before .Next().
bpmn_attrs()¶Extra XML attributes for the node’s BPMN element.
get_available_actions(activation: Activation, user: Any) Iterator[Tuple[str, str]]¶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).
has_view_permission(user: Any, obj: Any | None = None)¶Determine if the user has permission to view the viewset.
parent: BaseViewset | None¶Get the parent viewset.
reverse(viewname: str, args: List[Any] | None = None, kwargs: Dict[str, Any] | None = None, current_app: str | None = None) str¶Get the URL for a given viewname, including the namespace.