Simple Workflow Engine Library
Project description
Pieuvre
Pieuvre is a simple yet powerful workflow engine library developed by Kosc Telecom, aimed at Django but also usable in standalone mode.
Getting Started
Prerequisites
- Python 3.5+
- Optional: Django 1.11+
Installing
pip install pieuvre
Running the tests
Pieuvre's tests could be quite improved. However you can have a look! Install the dependencies to run the tests:
pip install pieuvre[test]
Then run them with:
pytest
Usage
Pieuvre allows you to attach workflows to backend models (built-in support for Django models, but any class implementing a save
method will work).
Pieuvre workflows define a set of states and transitions and allow quick implementation of custom hooks for each transition. Pieuvre lets you implement complex business logic backed by any storage implementation.
Custom behavior can be set:
- for transitions (by defining a method name with the transition name),
- when a state is entered (
on_enter_<state_name>
)`or exited (on_exit_<state_name>
) - before (
before_<transition_name>
) or after (after_<transition_name>
) a transition
Those hooks are dynamically called and require no specific setup.
Checks allow you to enforce specific conditions for entering or leaving a state.
Example:
from pieuvre import Workflow, WorkflowEnabled
ROCKET_STATES = Choices(
"IN_FACTORY", "in_factory", "in factory",
"ON_LAUNCHPAD", "on_launchpad", "on launchpad",
"IN_SPACE", "in_space", "in space"
)
ROCKET_BRANDS = Choices(
"ARIANESPACE", "arianespace", "Arianespace",
"SPACEX", "spacex", "Space X"
)
class Rocket(WorkflowEnabled, models.Model):
state = models.CharField(default=ROCKET_STATES.IN_FACTORY, choices=ROCKET_STATES)
fuel = models.PositiveIntegerField(default=0)
launch_date = models.DateTimeField(null=True)
brand = models.CharField()
def get_workflow_class(self):
if self.brand == ROCKET_BRANDS.ARIANESPACE:
return Ariane5Workflow()
return RocketWorkflow()
class RocketWorkflow(Workflow):
states = ROCKET_STATES
transitions = {
[
"source": ROCKET_STATES.IN_FACTORY,
"destination": ROCKET_STATES.ON_LAUNCHPAD,
"name": "prepare_for_launch"
],
[
"source": ROCKET_STATES.ON_LAUNCHPAD,
"destination": ROCKET_STATES.IN_SPACE,
"name": "launch"
]
}
def _refill(self):
self.fuel += 1000
def prepare_for_launch(self):
if self.model.fuel < 10:
self._refill()
def launch(self):
self.model.launch_date = timezone.now()
class Ariane5Workflow(RocketWorkflow):
@on_enter_state_check(ROCKET_STATES.IN_SPACE)
def check_launch(self):
if self.model.fuel < 220:
raise WorkflowValidationError("Not enough fuel to go up!")
if __name__ == "__main__":
rocket = Rocket.objects.create(brand=ROCKET_BRANDS.ARIANESPACE)
rocket.workflow.prepare_for_launch()
rocket.workflow.launch()
assert rocket.launch_date is not None
Workflows can be extended and dynamically instanciated. This lets you implement multiple workflows backed by a single model, which allows powerful business logic customization as well as a true split between the model definition and its behavior.
Workflows just need a field to store their state (state
by default, but easily overridable with state_field_name
). It is thus possible to let different workflows coexist on the same model, for instance a workflow modelizing the launch procedure of a rocket and an other workflow modelizing the launch in orbit of its payload.
Contributing
Any contribution is welcome through Github's Pull requests.
Ideas:
- store a workflow version to allow graceful workflow upgrades while maintaining workflow consistency on existing objects
- support for other ORM backends
Authors
- Saïd Ben Rjab - Kosc Telecom
- lerela - Fasfox
License
This project is licensed under the Apache License - see the LICENSE.md file for details
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.