core

Foxtrot core functionality

Workflow

 Workflow (config={})

Utility class for managing workflows with state transitions; built on pytransitions

source

Workflow.cache

 Workflow.cache (key)

Decorator for ensuring that tests (conditions/unless) only get run once; key should be set to a unique value per test.

This example configures a generic workflow which will always progress to “success” when next is called; note that there’s no need for caching because always and never are idempotent functions:

BORING_WORKFLOW = {
    "states": ("start", "success", "error"),
    "initial": "start",
    "transitions": (
        {"trigger": "next", "source": "start", "dest": "success", "conditions": "always", "unless": "never" },
        {"trigger": "next", "source": "start", "dest": "error", "conditions": "never", "unless": "always" },
    ),
    "auto_transitions": False,
    "send_event": True,
}

class BoringWorkflow(Workflow):

    def always(self, event_data):
        return True

    def never(self, event_data):
        return False

workflow=BoringWorkflow(BORING_WORKFLOW)
workflow.next()
print(f'Workflow state: {workflow.state}')

Workflow state: success

In this example we create a workflow that will succeed or fail based on the result of coinflip; here we use the @Workflow.cache decorator to preserve the results of coinflip, otherwise we would be resetting the value as each trigger is evaluated, leading to inconsistent results:

COINFLIP_WORKFLOW = {
    "states": ("start", "success", "error"),
    "initial": "start",
    "transitions": (
        {"trigger": "next", "source": "start", "dest": "error", "unless": "coinflip" },
        {"trigger": "next", "source": "start", "dest": "success", "conditions": "coinflip" },
    ),
    "auto_transitions": False,
    "send_event": True,
}

class CoinflipWorkflow(Workflow):

    @Workflow.cache('coinflip')
    def coinflip(self, event_data):
        return choice((True, False))

workflow=CoinflipWorkflow(COINFLIP_WORKFLOW)
workflow.next()
print(f'__cache__: {workflow.__cache__}\nWorkflow state: {workflow.state}')

__cache__: {'coinflip': False}
Workflow state: error