This is a pre-production deployment of Warehouse, however changes made here WILL affect the production instance of PyPI.
Latest Version Dependencies status unknown Test status unknown Test coverage unknown
Project Description

Framework/tools for python that get work done, make code easy to test,
and turn spaghetti logic back into something you can love.

The modeling here provides clear contracts, state encapsulation,
and strict typing.


You'll find the [full example](./tests/workflow/ in the tests/ directory.

This defines a workflow around the imperative: "Throw a Pie".

In a simple system this is about 5 lines of code. But in others,
where business logic requires the interaction of many systems,
logic is smeared around and duplicated, and not well encapsulated.

The approach here splits execution into discrete units which can be
composed into other workflows encouraging reuse, testability, etc.

from marx.workflow.step import LogicUnit, ArgSpec, Step, ResultSpec
from marx.workflow.flow import Workflow
from marx.workflow.context import DefaultContext, Field
from tests.workflow.example_objects import User, PermissionDeniedError

class IsUserAuthorized(LogicUnit):
"""Checks permission for a user+action, and notifies authorities
if it fails."""

user = ArgSpec(User, docs="The user performing the action")

def __init__(self, action):
@param action: The action that will be checked.
self.action = action

def __call__(self, user):
@param user: The user performing the action.
if self.is_authorized(user):
self.notify_authorities(user, self.action)
raise PermissionDeniedError(self.action)

def is_authorized(self, user):
return in ("bob", "mary")

def notify_authorities(self, user, action):
print "AUTHORITIES!!!", user, " attempted illegal action", self.action

class MakePie(LogicUnit):
"""Makes the pie."""

maker = ArgSpec(User, docs="The person making pie.")
pie = ResultSpec(basestring, docs="Kind of pie")

def __call__(self, maker):
maker.increment("pies_made", 1)
self.result.pie = 'lemon'

class ThrowThing(LogicUnit):
"""Subject Object (Verb) Indirect-object"""

actor = ArgSpec(User)
hit = ResultSpec(bool, default=False, docs="Did we get 'em?")

# we omit target and thing here, because we don't
# need to enumerate/type constrain the values in this example

def __call__(self, actor, thing, target):
print "Throwing", thing
self.result.hit = actor.can_throw()
# we don't need to return, but we can.
return self.result

class ThrowPieContext(DefaultContext):
"""The execution context for the ThrowPieWorkflow."""
thrower = Field(User, docs="Somebody has to throw it")
target = Field(User, docs="At somebody")
pie = Field(str, docs="A pie, which we make along the way")
was_hit = Field(bool, docs="Success of the throwing event")

""" A workflow is a series of steps."""
IsUserAuthorizedStep = Step(
# we bind from the context to the arguments of the method.
arg_map={IsUserAuthorized.USER: ThrowPieContext.THROWER}
MakePieStep = Step(
arg_map={MakePie.MAKER: ThrowPieContext.THROWER},
# we bind from the returned result back to the context
ThrowThingStep = Step(
arg_map=ThrowThing.AutoMap({ThrowThing.ACTOR: ThrowPieContext.THROWER,
ThrowThing.THING: ThrowPieContext.PIE}),
result_map={ThrowPieContext.WAS_HIT: 'hit'}

""" There are a few ways to build up a workflow. By constructor..."""
ThrowPieWorkflowA = Workflow(steps=[

""" Or using add_step..."""
ThrowPieWorkflowB = Workflow().add_step(

""" Or using the overloaded addition operator."""
EmptyWorkflow = Workflow()
ThrowPieWorkflowC_A = EmptyWorkflow + IsUserAuthorizedStep + MakePieStep
ThrowPieWorkflowC_B = EmptyWorkflow + ThrowThingStep
ThrowPieWorkflowC = ThrowPieWorkflowC_A + ThrowPieWorkflowC_B

def run():
"""To execute a workflow, prepare a context, and pass it through."""
ctx = ThrowPieContext()
ctx.thrower = User("bob") = User("frank")
for WorkflowType in (ThrowPieWorkflowA,
assert ctx.was_hit is not None
assert ctx.pie == 'lemon'
return ctx
except PermissionDeniedError:
assert False

# Ensure that our ThrowPieWorkflowC did not modify its components
assert EmptyWorkflow.steps == []
assert ThrowThingStep not in ThrowPieWorkflowC_A.steps
Release History

Release History


This version

History Node

TODO: Figure out how to actually get changelog content.

Changelog content for this version goes here.

Donec et mollis dolor. Praesent et diam eget libero egestas mattis sit amet vitae augue. Nam tincidunt congue enim, ut porta lorem lacinia consectetur. Donec ut libero sed arcu vehicula ultricies a non tortor. Lorem ipsum dolor sit amet, consectetur adipiscing elit.

Show More

Download Files

Download Files

TODO: Brief introduction on what you do with files - including link to relevant help section.

File Name & Checksum SHA256 Checksum Help Version File Type Upload Date
marx-workflows-0.0.8.tar.gz (19.6 kB) Copy SHA256 Checksum SHA256 Source Oct 25, 2014

Supported By

WebFaction WebFaction Technical Writing Elastic Elastic Search Pingdom Pingdom Monitoring Dyn Dyn DNS Sentry Sentry Error Logging CloudAMQP CloudAMQP RabbitMQ Heroku Heroku PaaS Kabu Creative Kabu Creative UX & Design Fastly Fastly CDN DigiCert DigiCert EV Certificate Rackspace Rackspace Cloud Servers DreamHost DreamHost Log Hosting