Viewflow - Reusable Workflow Library

            Open-source
            
            Private

                Feel free to
                ask a question on Stackoverflow. Don't forget to add
                the [django-viewflow]
                tag, to get us notified.
            
                Pro customers email support contact is
                support@viewflow.zendesk.com
                or use green button at the bottom right corner.

Contacts

Mikhail Podgurskiy

Software developer

Came to idea of workflow library when fighting with Workflows GUI in 2004
Spent 5 years in the banking industry to shoulder-to-shoulder work with business on automation.
Wrote open sourced django-fsm library to reduce pain of business apps development at 2010

MODELS.PY

Code

from django.db import models
from viewflow.models import Process, Task


class Carrier(models.Model):
    DEFAULT = 'Default'

    name = models.CharField(max_length=50)
    phone = models.CharField(max_length=15)

    def is_default(self):
        return self.name == Carrier.DEFAULT

    def __str__(self):
        return self.name


class Insurance(models.Model):
    company_name = models.CharField(max_length=50)
    cost = models.IntegerField()

    def __str__(self):
        return '{} ${}'.format(self.company_name, self.cost)


class Shipment(models.Model):
    shipment_no = models.CharField(max_length=50)
    carrier = models.ForeignKey(Carrier, null=True)

    # customer
    first_name = models.CharField(max_length=150)
    last_name = models.CharField(max_length=150)
    email = models.EmailField()

    # shipment addres
    address = models.CharField(max_length=150)
    city = models.CharField(max_length=150)
    state = models.CharField(max_length=150)
    zipcode = models.CharField(max_length=10)
    country = models.CharField(max_length=150)
    phone = models.CharField(max_length=50)

    # shipment data
    need_insurance = models.BooleanField(default=False)
    insurance = models.ForeignKey('Insurance', null=True)

    carrier_quote = models.IntegerField(default=0)
    post_label = models.TextField(blank=True, null=True)

    package_tag = models.CharField(max_length=50)


class ShipmentItem(models.Model):
    shipment = models.ForeignKey(Shipment)
    name = models.CharField(max_length=250)
    quantity = models.IntegerField(default=1)


class ShipmentProcess(Process):
    shipment = models.ForeignKey(Shipment, blank=True, null=True)

    def is_normal_post(self):
        try:
            if self.shipment.carrier:
                return self.shipment.carrier.is_default()
            else:
                return None
        except (Shipment.DoesNotExist, Carrier.DoesNotExist):
            return None

    def need_extra_insurance(self):
        try:
            return self.shipment.need_insurance
        except Shipment.DoesNotExist:
            return None

    class Meta:
        permissions = [  # Explicitly specified, but also can be autogenerated by Viewflow
            ('can_start_request', 'Can start shipment request'),
            ('can_take_extra_insurance', 'Can take extra insurance'),
            ('can_package_goods', 'Can package goods'),
            ('can_move_package', 'Can move package')
        ]

VIEWS.PY

Code

from django.views import generic
from viewflow.flow.views import StartFlowMixin, FlowViewMixin

from .forms import ShipmentForm
from .models import Insurance


class StartView(StartFlowMixin, generic.UpdateView):
    form_class = ShipmentForm

    def get_object(self):
        return self.activation.process.shipment

    def activation_done(self, form):
        shipment = form.save()
        self.activation.process.shipment = shipment
        super(StartView, self).activation_done(form)


class ShipmentView(FlowViewMixin, generic.UpdateView):
    def get_object(self):
        return self.activation.process.shipment


class InsuranceView(FlowViewMixin, generic.CreateView):
    model = Insurance
    fields = ['company_name', 'cost']

    def activation_done(self, form):
        shipment = self.activation.process.shipment
        shipment.insurance = self.object
        shipment.save(update_fields=['insurance'])
        self.activation.done()

FLOWS.PY

Code

from viewflow import flow
from viewflow.base import this, Flow
from viewflow.flow.views import UpdateProcessView
from viewflow.lock import select_for_update_lock

from . import views
from .models import ShipmentProcess


class ShipmentFlow(Flow):
    """
    Shipment

    Shipment workflow for e-commerce store back-office automation
    """
    process_class = ShipmentProcess
    lock_impl = select_for_update_lock

    summary_template = """
        Shipment  items
        to   / 
        """

    start = (
        flow.Start(views.StartView)
        .Permission('shipment.can_start_request')
        .Next(this.split_clerk_warehouse)
    )

    # clerk
    split_clerk_warehouse = (
        flow.Split()
        .Next(this.shipment_type)
        .Next(this.package_goods)
    )

    shipment_type = (
        flow.View(
            views.ShipmentView, fields=["carrier"],
            task_description="Carrier selection")
        .Assign(lambda act: act.process.created_by)
        .Next(this.delivery_mode)
    )

    delivery_mode = (
        flow.If(cond=lambda act: act.process.is_normal_post())
        .Then(this.check_insurance)
        .Else(this.request_quotes)
    )

    request_quotes = (
        flow.View(
            views.ShipmentView,
            fields=["carrier_quote"])
        .Assign(lambda act: act.process.created_by)
        .Next(this.join_clerk_warehouse)
    )

    check_insurance = (
        flow.View(
            views.ShipmentView,
            fields=["need_insurance"])
        .Assign(lambda act: act.process.created_by)
        .Next('split_on_insurance')
    )

    split_on_insurance = (
        flow.Split()
        .Next(
            this.take_extra_insurance,
            cond=lambda act: act.process.need_extra_insurance())
        .Always(this.fill_post_label)
    )

    fill_post_label = (
        flow.View(
            views.ShipmentView,
            fields=["post_label"])
        .Assign(lambda act: act.process.created_by)
        .Next(this.join_on_insurance)
    )

    join_on_insurance = (
        flow.Join()
        .Next(this.join_clerk_warehouse)
    )

    # Logistic manager
    take_extra_insurance = (
        flow.View(views.InsuranceView)
        .Permission('shipment.can_take_extra_insurance')
        .Next(this.join_on_insurance)
    )

    # Warehouse worker
    package_goods = (
        flow.View(UpdateProcessView)
        .Permission('shipment.can_package_goods')
        .Next(this.join_clerk_warehouse)
    )

    join_clerk_warehouse = (
        flow.Join()
        .Next(this.move_package)
    )

    move_package = (
        flow.View(UpdateProcessView)
        .Assign(this.package_goods.owner)
        .Next(this.end)
    )

    end = flow.End()

URLS.PY

Code

from viewflow.flow.viewset import FlowViewSet
from .flows import ShipmentFlow

urlpatterns = [
    url(r'^shipment', include(FlowViewSet(ShipmentFlow).urls, namespace='shipment'))
]

Getting help

Open-source

Private

Contacts

MODELS.PY

Code

VIEWS.PY

Code

FLOWS.PY

Code

URLS.PY

Code