Как программно создать процесс django ViewFlow

Краткое содержание

Я разрабатываю веб-приложение для изучения Django (python 3.4 и Django 1.6.10). Веб-приложение имеет сложные и часто обновляемые рабочие процессы. Я решил интегрировать библиотеку 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 /flow.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 содержит правильный метод, но не уверен, как все должно быть организовано.

Заранее спасибо!


python django workflow django-viewflow
person Melek Somai    schedule 26.02.2015    source источник

Ответы (3)


arrow_upward
12
arrow_downward

Для Flows доступны две дополнительные встроенные задачи Start.

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.

Чтобы вручную обработать состояние задачи, сначала вы должны получить задачу из базы данных, активировать ее и вызвать любой метод активации.

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

Любой переход активации имеет .can_proceed() метод, помогающий вам проверить, находится ли задача в состоянии, которое разрешает переход.

person kmmbvnr    schedule 26.02.2015
comment
Ссылка на github дает 404, FWIW - person Bryan Oakley; 26.05.2016
comment
AttributeError: объект 'StartFunction' не имеет атрибута 'prepare' - person Narnik Gamarnik; 08.04.2017
comment
Да, в последней версии дополнительно требуется декоратор @flow_start_func. - person kmmbvnr; 08.04.2017
comment
@kmmbvnr Могу ли я использовать сигнал post_save модели django с StartSignal и как его реализовать? Я разместил свой вопрос здесь: stackoverflow.com/questions/53980834/ - person Silko; 30.12.2018

arrow_upward
1
arrow_downward

Мне нужно было запускать экземпляр потока вручную или программно. Модель, к которой я пришел, на основе приведенной выше ссылки на 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

Обратите внимание, что подкласс activation внутри flow_start_func - это FuncActivation, который имеет методы prepare () и save (). kwargs происходит от вызова run, который выглядит примерно так:

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

arrow_upward
0
arrow_downward

для запуска вручную:

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()
person Luis Quissak    schedule 31.05.2019