如何以编程方式创建 django ViewFlow 进程

Question

剧情简介

我正在开发一个网络应用程序来学习 Django（python 3.4 和 Django 1.6.10）。 Web 应用程序具有复杂且经常更新的工作流程。我决定集成 Django-Viewflow 库 (https://github.com/viewflow/viewflow/)，因为它似乎是处理工作流的一种非常方便的方式，而不是将工作流逻辑与应用程序模型结合起来。

在这种情况下，我创建了一个工作流来使用 Django-Viewflow 库收集作者身份信息和版权。每次将作者添加到图书时，都应启动工作流程。

我的问题

该文档提供了集成端到端工作流解决方案（前端和后端）的分步指南。我的问题是我很难以编程方式控制工作流（特别是从 Book 模型）。

应用说明

我有一个与作者有多对多关系的书籍模型（核心模型）。

myApp/models.py

class Book(models.Model):
    title = models.CharField(max_length=100)
    authors = models.ManyToManyField(Author)
    publisher = models.ForeignKey(Publisher)
    publication_date = models.DateField()

工作流组件是：

myFlow/models.py

from viewflow.models import Process

class AuthorInvitation(process)     
    consent_confirmed = models.BooleanField(default=False)
    signature = models.CharField(max_length=150)

myFlow/flows.py

from viewflow import flow
from viewflow.base import this, Flow
from viewflow.contrib import celery
from viewflow.views import StartProcessView, ProcessView
from . import models, tasks

class AuthorInvitationFlow(Flow):
    process_cls = models.AuthorInvitation

    start = flow.Start(StartProcessView) \
        .Permission(auto_create=True) \
        .Next(this.notify)

    notify = celery.Job(tasks.send_authorship_request) \
        .Next(this.approve)

    approve = flow.View(ProcessView, fields=["confirmed","signature"]) \
        .Permission(auto_create=True) \
        .Next(this.check_approve)

    check_approve = flow.If(cond=lambda p: p.confirmed) \
        .OnTrue(this.send) \
        .OnFalse(this.end)

    send = celery.Job(tasks.send_authorship) \
        .Next(this.end)

    end = flow.End()

问题

如何以编程方式控制工作流过程（激活、确认步骤、重做步骤、取消过程......）？我试图深入研究图书馆的代码。 class activate 似乎包含正确的方法，但不确定整体应该如何编排。

提前致谢！

Answer 1

还有两个可用于流的额外启动内置任务

StartFunction - 当函数在某处调用时开始流程：

@flow_start_func
def create_flow(activation, **kwargs):
    activation.prepare()
    activation.done()
    return activation

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

# somewhere in the code
FunctionFlow.start.run(**some_kwargs)

StartSignal - 在接收到 Django 信号时启动流程：

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

您可以在 this viewflow test suite.

中查看它们的用法，以及其余的内置任务

对于手动处理任务状态，首先你应该从数据库中获取任务，激活它，然后调用任何激活方法。

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

任何激活转换都有.can_proceed()方法，帮你检查一下，是不是处于允许转换状态的任务。

Answer 2

我需要能够手动或以编程方式启动流实例。基于上面对 StartFunction 的引用，我最终得到的模型如下所示：

class MyRunFlow(flow.Flow):
    process_class = Run

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

请注意，重要的一点是 process_create 具有 Process 对象，并且此代码必须以编程方式设置与手动表单提交通过 ProcessCreate 的字段规范所做的相同的字段:

@flow_start_func
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
    activation.process.save()
    #
    # Go!
    #
    activation.prepare()
    activation.done()
    return activation

注意flow_start_func里面的activation子类是FuncActivation，它有prepare()和save()方法。 kwargs 来自对运行的调用，类似于：

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

Answer 3

手动启动：

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

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

    @method_decorator(flow.flow_start_func)
    def create_flow(self, activation):
        activation.prepare()
        activation.done()
        return activation

之后公开 rpc:

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

    @rpc_method
    def start():
        CIE.start.run()

如何以编程方式创建 django ViewFlow 进程

How to create a django ViewFlow process programmatically

python

django

workflow

django-viewflow