[Fixed]-How to create a django ViewFlow process programmatically


There are two additional Start build-in Tasks available for Flows

StartFunction – starts flow when the function called somewhere:

def create_flow(activation, **kwargs):
    return activation

class FunctionFlow(Flow):
    start = flow.StartFunction(create_flow) \

# somewhere in the code

StartSignal – starts flow on django signal receive:

class SignalFlow(Flow):
    start = flow.StartSignal(some_signal, create_flow) \      

You can check the usage for them, and rest of build-in task in this viewflow test suite.

For manually process the task state, first you should get the task from the database, activate it, and call any activation method.

task  = MyFlow.task_cls.objects.get(...)
activation = task.activate()
if  activation.undo.can_proceed():

Any activation transition have .can_proceed() method, helps you to check, is the task in the state that allows the transition.


I needed to be able to manually or programmatically start a flow instance. The model I ended up with, based on the above reference to StartFunction looks like this:

class MyRunFlow(flow.Flow):
    process_class = Run

    start = flow.Start(ProcessCreate, fields=['schedule']). \
        Permission(auto_create=True). \
    start2 = flow.StartFunction(process_create). \

Note the important point is that process_create has the Process object and this code must programmatically set up the same fields that the manual form submission does via the fields specification to ProcessCreate:

def process_create(activation: FuncActivation, **kwargs):
    # Update the database record.
    db_sch = Schedule.objects.get(id=kwargs['schedule'])
    activation.process.schedule = db_sch # <<<< Same fields as ProcessCreate
    # Go!
    return activation

Note that the activation subclass inside the flow_start_func is FuncActivation, which has the prepare() and save() methods. The kwargs come from the call to run, which goes something like:

start_node = <walk the flow class looking for nodes of type StartFunction>
activation = start_node.run(schedule=self.id)


to start manually:

from viewflow import flow
from viewflow.base import this, Flow
from .models import CIE
from viewflow import frontend
from django.utils.decorators import method_decorator

class CIE(Flow):
    process_class = CIE
    start = flow.StartFunction(this.create_flow).Next(this.end)
    end = flow.End()

    def create_flow(self, activation):
        return activation

after that expose rpc:

from modernrpc.core import rpc_method
from flow.flows import CIE

    def start():

Leave a comment