SpiffWorkflow

https://spiffworkflow.readthedocs.io/en/latest/
Admin
SpiffWorkflow is a flexible Python library that provides a workflow execution environment with...
Tokens:35,570
Snippets:476
Trust Score:9.6
Update:5 months ago
Show doc for...
Tokens:
Raw
### Running Camunda Example

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/camunda/index.rst.txt

Use this command to run the Camunda example with a SQLite database.

```console
./runner.py -e spiff_example.camunda.sqlite
```

--------------------------------

### Loading Example Data with Runner Script

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/data.rst.txt

Use this command to load the example data for the 'order_product' process, including BPMN and DMN files.

```console
./runner.py -e spiff_example.spiff.file add -p order_product \
    -b bpmn/tutorial/{top_level,data_output}.bpmn \
    -d bpmn/tutorial/{shipping_costs,product_prices}.dmn
```

--------------------------------

### Run the threaded service task example

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

Commands to add and then run the threaded service task example using the `runner.py` script. This involves specifying the example module and the BPMN file.

```bash
./runner.py -e spiff_example.misc.threaded_service_task add -p threaded_service -b bpmn/tutorial/threaded_service_task.bpmn
./runner.py -e spiff_example.misc.threaded_service_task

```

--------------------------------

### Run threaded service task example

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Command-line instructions to add and run the threaded service task example using the runner script. It specifies the example module and the BPMN file.

```console
./runner.py -e spiff_example.misc.threaded_service_task add -p threaded_service -b bpmn/tutorial/threaded_service_task.bpmn
./runner.py -e spiff_example.misc.threaded_service_task
```

--------------------------------

### Execute Example Commands

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Demonstrates how to run BPMN processes using the SpiffWorkflow runner script. Includes examples for adding a process and running with different engine configurations.

```console
./runner.py -e spiff_example.spiff.file add -p end_it_all -b bpmn/tutorial/dangerous.bpmn
```

```console
./runner.py -e spiff_example.spiff.file
```

--------------------------------

### Console Command to Load Example

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

This is a placeholder for a console command that would be used to load or run the described example. The actual command is not provided in the source.

```console

```

--------------------------------

### Run Example with Camunda SQLite

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/camunda/index.html

Execute a SpiffWorkflow example using the Camunda configuration with a SQLite database.

```bash
./runner.py -e spiff_example.camunda.sqlite

```

--------------------------------

### Runner Command for Custom Exec Example

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Example console commands to run the custom execution example using the runner script. These commands specify the execution module, arguments, and BPMN files to process.

```console
./runner.py -e spiff_example.spiff.custom_exec add -p order_product \
    -b bpmn/tutorial/{top_level_script,call_activity_script}.bpmn
./runner.py -e spiff_example.spiff.custom_exec
```

--------------------------------

### Load Example with Runner Script

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

Command to load and run the event handling example using the runner script. This command specifies the Python module, the action to add, and the BPMN file to use.

```bash
./runner.py -e spiff_example.misc.event_handler add -p read_file -b bpmn/tutorial/event_handler.bpmn
```

--------------------------------

### Run Custom Task Spec Example

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/custom_task_spec.html

Execute a workflow with a custom task spec using the runner script. This command adds the timer_start process and then runs it.

```bash
./runner.py -e spiff_example.misc.custom_start_event add -p timer_start -b bpmn/tutorial/timer_start.bpmn
./runner.py -e spiff_example.misc.custom_start_event

```

--------------------------------

### Create and Run Workflow from JSON Specification

Source: https://spiffworkflow.readthedocs.io/en/latest/core/tutorial/index.html

Load a workflow specification from a JSON file, create a Workflow instance, and execute it. This example assumes the JSON file is named 'nuclear.json'.

```python
from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from SpiffWorkflow.serializer.json import JSONSerializer

# Load from JSON
with open('nuclear.json') as fp:
    workflow_json = fp.read()
serializer = JSONSerializer()
spec = WorkflowSpec.deserialize(serializer, workflow_json)

# Alternatively, create an instance of the Python based specification.
#from nuclear import NuclearStrikeWorkflowSpec
#spec = NuclearStrikeWorkflowSpec()

# Create the workflow.
workflow = Workflow(spec)

# Execute until all tasks are done or require manual intervention.
# For the sake of this tutorial, we ignore the "manual" flag on the
# tasks. In practice, you probably don't want to do that.
workflow.run_all(halt_on_manual=False)

# Alternatively, this is what a UI would do for a manual task.
#workflow.complete_task_from_id(...)

```

--------------------------------

### Initialize Serializer Directory

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/application.rst.txt

Define the directory name where serialized workflow data will be stored. This is a common setup step for serializers.

```python
dirname = 'wfdata'
```

--------------------------------

### Add Custom Services to Scripting Environment

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Make custom functions like product info lookup and shipping cost calculation available to the scripting environment. This example also includes adding the 'datetime' module.

```python
script_env = TaskDataEnvironment({
        'datetime': datetime,
        'lookup_product_info': lookup_product_info,
        'lookup_shipping_cost': lookup_shipping_cost,
    })
```

--------------------------------

### Use Custom Serializer and Task in Practice

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/core/custom-tasks/index.rst.txt

Instantiate the custom serializer and use it to load a workflow specification that includes the custom task. Then, start and run the workflow.

```python
from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.bpmn.workflow import BPMNWorkflow
from SpiffWorkflow.serializer.json import JSONSerializer

# Assuming strike.py and serializer.py are in the same directory
from strike import NuclearStrike
from serializer import CustomSerializer

# Load the workflow specification from JSON
workflow_spec_json = {
    "name": "nuclear_strike_workflow",
    "tasks": [
        {
            "name": "launch_rocket",
            "type": "strike.NuclearStrike"
        }
    ]
}

# Use the custom serializer
serializer = CustomSerializer()
workflow_spec = serializer.deserialize_workflow_spec(workflow_spec_json)

# Create a workflow instance
workflow = Workflow(workflow_spec)

# Start the workflow
workflow.start()

# Get the current task
current_task = workflow.get_current_tasks()[0]

# Complete the task
current_task.complete()

# Check if workflow is finished
print(f"Workflow finished: {workflow.is_finished()}")
```

--------------------------------

### Example User Input Data

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Represents the data structure collected after a user submits a form for a product and quantity selection.

```json
{
    'product_name': 'product_a',
    'product_quantity': 2,
}
```

--------------------------------

### Instantiate a BPMN Workflow

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/workflows.html

Starts a new workflow instance from a given specification ID. It retrieves the workflow specification and creates a new workflow object, returning its ID and the workflow instance.

```python
def start_workflow(self, spec_id):
    spec, sp_specs = self.serializer.get_workflow_spec(spec_id)
    wf = BpmnWorkflow(spec, sp_specs, script_engine=self._script_engine)
    wf_id = self.serializer.create_workflow(wf, spec_id)
    return Instance(wf_id, workflow)
```

--------------------------------

### Create Custom Start Event Converter

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/custom_task_spec.html

Implement a converter for the custom task spec to enable serialization. This class handles the conversion of event definitions to and from dictionaries.

```python
from SpiffWorkflow.bpmn.serializer import BpmnWorkflowSerializer
from SpiffWorkflow.spiff.serializer.task_spec import SpiffBpmnTaskConverter
from SpiffWorkflow.spiff.serializer import DEFAULT_CONFIG

class CustomStartEventConverter(SpiffBpmnTaskConverter):

    def to_dict(self, spec):
        dct = super().to_dict(spec)
        dct['event_definition'] = self.registry.convert(spec.event_definition)
        dct['timer_event'] = self.registry.convert(spec.timer_event)
        return dct

    def from_dict(self, dct):
        spec = super().from_dict(dct)
        spec.event_definition = self.registry.restore(dct['event_definition'])
        spec.timer_event = self.registry.restore(dct['timer_event'])
        return spec

```

--------------------------------

### Custom Service Task Environment

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Extend TaskDataEnvironment to provide custom service implementations. This example shows how to implement a 'read_file' operation that reads file content or raises FileNotFoundError.

```python
class ServiceTaskEnvironment(TaskDataEnvironment):

    def call_service(self, context, operation_name, operation_params):
        if operation_name == 'read_file':
            return open(operation_params['filename']).read()
        else:
            raise ValueError('Unknown Service')
```

--------------------------------

### Get DMN Dependencies

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/parsing.html

Obtain a list of DMN IDs that are referenced by a given process ID.

```python
get_dmn_dependencies
```

--------------------------------

### Initialize RestrictedPython Environment

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Replace the default global environment with one provided by RestrictedPython for security. This setup is useful when you need to limit the script engine's capabilities.

```python
from RestrictedPython import safe_globals
from SpiffWorkflow.bpmn.script_engine import TaskDataEnvironment

script_env = TaskDataEnvironment(safe_globals)
```

--------------------------------

### Adding Service Task Workflow via Runner

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

Command-line instruction to add a workflow that includes a service task. This command uses the `runner.py` script to add a specific example and BPMN files.

```bash
./runner.py -e spiff_example.spiff.service_task add -p order_product \
    -b bpmn/tutorial/{top_level_service_task,call_activity_service_task}.bpmn

```

--------------------------------

### Service Task Configuration in BPMN

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Example of a BPMN Service Task configuration using `spiffworkflow` extensions. It defines the service operator, parameters, and a post-script to process the result.

```xml
<bpmn:serviceTask id="Activity_1ln3xkw" name="Lookup Product Info">
      <bpmn:extensionElements>
        <spiffworkflow:serviceTaskOperator id="lookup_product_info" resultVariable="product_info">
          <spiffworkflow:parameters>
            <spiffworkflow:parameter id="product_name" type="str" value="product_name"/>
          </spiffworkflow:parameters>
        </spiffworkflow:serviceTaskOperator>
        <spiffworkflow:postScript>product_info = product_info_from_dict(product_info)</spiffworkflow:postScript>
      </bpmn:extensionElements>
      <bpmn:incoming>Flow_104dmrv</bpmn:incoming>
      <bpmn:outgoing>Flow_06k811b</bpmn:outgoing>
    </bpmn:serviceTask>
```

--------------------------------

### Create Custom Start Event Converter

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/custom_task_spec.rst.txt

Implement a converter for the custom task spec to enable serialization and deserialization. This class inherits from SpiffBpmnTaskConverter and handles custom attributes like event definitions.

```python
from SpiffWorkflow.bpmn.serializer import BpmnWorkflowSerializer
from SpiffWorkflow.spiff.serializer.task_spec import SpiffBpmnTaskConverter
from SpiffWorkflow.spiff.serializer import DEFAULT_CONFIG

class CustomStartEventConverter(SpiffBpmnTaskConverter):

    def to_dict(self, spec):
        dct = super().to_dict(spec)
        dct['event_definition'] = self.registry.convert(spec.event_definition)
        dct['timer_event'] = self.registry.convert(spec.timer_event)
        return dct

    def from_dict(self, dct):
        spec = super().from_dict(dct)
        spec.event_definition = self.registry.restore(dct['event_definition'])
        spec.timer_event = self.registry.restore(dct['timer_event'])
        return spec
```

--------------------------------

### Define a Custom Task

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/core/custom-tasks/index.rst.txt

Extend `SpiffWorkflow.specs.Simple` to create a custom task. Implement `_on_complete_hook` for custom logic when the task completes. This example defines a `NuclearStrike` task.

```python
from SpiffWorkflow.specs import Simple

class NuclearStrike(Simple):
    def _on_complete_hook(self, my_task):
        print("Rocket sent!")
```

--------------------------------

### Get Next Task and Subprocess

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Retrieve the next task matching a specific name and then get the associated subprocess for that task.

```python
task = workflow.get_next_task(spec_name='customize_product')
subprocess = workflow.get_subprocess(task)
```

--------------------------------

### Define Custom Start Event Spec

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/custom_task_spec.html

Create a custom task spec for handling Timer Start Events. This class replaces TimerEventDefinition with NoneEventDefinition for external management.

```python
from SpiffWorkflow.bpmn.specs.event_definitions import NoneEventDefinition
from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition
from SpiffWorkflow.bpmn.specs.mixins import StartEventMixin
from SpiffWorkflow.spiff.specs import SpiffBpmnTask

class CustomStartEvent(StartEventMixin, SpiffBpmnTask):

    def __init__(self, wf_spec, bpmn_id, event_definition, **kwargs):

        if isinstance(event_definition, TimerEventDefinition):
            super().__init__(wf_spec, bpmn_id, NoneEventDefinition(), **kwargs)
            self.timer_event = event_definition
        else:
            super().__init__(wf_spec, bpmn_id, event_definition, **kwargs)
            self.timer_event = None

```

--------------------------------

### Create and Execute Workflow

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/core/tutorial/index.rst.txt

Instantiate and run a workflow from a specification. Assumes all tasks are automatic for tutorial purposes.

```python
from spiffworkflow.workflow import Workflow
from spiffworkflow.specs import WorkflowSpec

# Assuming nuclear.py or nuclear.json has been loaded into spec
# For demonstration, let's use the Python-defined spec:
def create_nuclear_strike_workflow_spec():
    spec = WorkflowSpec()
    start_task = Task(name='start', spec=spec)
    spec.add_task(start_task)
    president_approval = Task(name='president_approval', spec=spec, manual=True)
    spec.add_task(president_approval)
    spec.add_transition(start_task, president_approval)
    general_approval = Task(name='general_approval', spec=spec, manual=True)
    spec.add_task(general_approval)
    spec.add_transition(president_approval, general_approval)
    launch_task = Task(name='launch', spec=spec)
    spec.add_task(launch_task)
    spec.add_transition(general_approval, launch_task)
    end_task = Task(name='end', spec=spec)
    spec.add_task(end_task)
    spec.add_transition(launch_task, end_task)
    return spec

spec = create_nuclear_strike_workflow_spec()
workflow = Workflow(spec)

# Simulate user input for manual tasks
# In a real UI, this would come from user interaction
workflow.task_events.on('task.waiting', lambda task: task.update_state(Task.STATE.READY))

# Complete all tasks until no more are ready
workflow.complete_all()

print(f"Workflow finished: {workflow.is_completed}")
print(f"Current task: {workflow.current_task.name if workflow.current_task else 'None'}")
```

--------------------------------

### Create BpmnEngine

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/application.html

Creates a BpmnEngine instance by passing the configured parser, serializer, and script environment. Handlers are automatically passed to the UI by the main runner.

```python
from ..engine import BpmnEngine
engine = BpmnEngine(parser, serializer, script_env)
```

--------------------------------

### Get Task Display Information

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Retrieves display-relevant information for a given task, including its state, name, and lane.

```python
def get_task_display_info(self, task):
        return {
            'depth': task.depth,
            'state': TaskState.get_name(task.state),
            'name': task.task_spec.bpmn_name or task.task_spec.name,
            'lane': task.task_spec.lane,
        }
```

--------------------------------

### Set User Task Instructions

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Renders Jinja template instructions for user tasks, incorporating workflow data. Requires Jinja2 and access to task data.

```python
from jinja2 import Template

def set_instructions(self, task):
    user_input = self.ui._states['user_input']
    user_input.instructions = f'{self.task.task_spec.bpmn_name}\n\n'
    text = self.task.task_spec.extensions.get('instructionsForEndUser')
    if text is not None:
        template = Template(text)
        user_input.instructions += template.render(self.task.data)
    user_input.instructions += '\n\n'
```

--------------------------------

### Get Process Dependencies

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/parsing.html

Obtain a list of process IDs that are referenced by a given process ID, typically through Call Activities or Subprocesses.

```python
get_process_dependencies
```

--------------------------------

### Get BPMN Process Specification

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/parsing.rst.txt

Retrieves the BpmnProcessSpec for a given process ID. This method also fetches associated subprocess specifications.

```python
spec = self.parser.get_spec(process_id)
dependencies = self.parser.get_subprocess_specs(process_id)
```

--------------------------------

### Import BpmnTaskSpec and TaskSpecMixin

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import BpmnTaskSpec for generic BPMN behavior and TaskSpecMixin for specific BPMN behavior extensions.

```python
from SpiffWorkflow.bpmn.specs import BpmnTaskSpec           # Implements generic BPMN behavior
from SpiffWorkflow.bpmn.specs.mixins import <TaskSpecMixin> # Implements specific BPMN behavior
```

--------------------------------

### List Workflow Instances

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/diffs.html

Use this command to list all running workflow instances. It displays instance IDs, workflow names, and creation/update timestamps.

```bash
./runner.py -e spiff_example.spiff.diffs list_instances

4af0e043-6fd6-448d-85eb-d4e86067433e  order_product        2024-07-02 17:46:57 2024-07-02 17:47:00
af180ef6-0437-41fe-b745-8ec4084f3c57  order_product        2024-07-02 17:47:05 2024-07-02 17:47:30

```

--------------------------------

### Get Tasks from Subworkflow

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Retrieves all tasks within a specific subworkflow. Subworkflows share the top-level workflow's script engine for consistency.

```python
subprocess_tasks = subprocess.get_tasks()
```

--------------------------------

### Get Workflow and Spec Diff

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Retrieves the current workflow instance and its specification, then calculates the difference between them. This function is used internally by the diffing process.

```python
def diff_workflow(self, wf_id, spec_id):
    wf = self.serializer.get_workflow(wf_id)
    spec, deps = self.serializer.get_workflow_spec(spec_id)
    return diff_workflow(self.serializer.registry, wf, spec, deps)
```

--------------------------------

### Implement BpmnDataStoreSpecification

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import BpmnDataStoreSpecification for implementing custom datastore specifications within BPMN workflows.

```python
from SpiffWorkflow.bpmn.spec import BpmnDataStoreSpecification
```

--------------------------------

### Using Custom Serializer and Task (Python)

Source: https://spiffworkflow.readthedocs.io/en/latest/core/custom-tasks/index.html

Demonstrates loading a workflow specification from a JSON file using a custom serializer and then executing the workflow. Ensure the 'nuclear.json' file exists and the custom serializer is correctly implemented.

```python
from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from serializer import NuclearSerializer

# Load from JSON
with open('nuclear.json') as fp:
    workflow_json = fp.read()
nuclear_serializer = NuclearSerializer()
spec = WorkflowSpec.deserialize(nuclear_serializer, workflow_json)

# Create the workflow.
workflow = Workflow(spec)

# Execute until all tasks are done or require manual intervention.
# For the sake of this tutorial, we ignore the "manual" flag on the
# tasks. In practice, you probably don't want to do that.
workflow.run_all(halt_on_manual=False)

```

--------------------------------

### Import BpmnWorkflowSerializer

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import BpmnWorkflowSerializer for basic serialization of BPMN workflows.

```python
from SpiffWorkflow.bpmn.serializer import BpmnWorkflowSerializer
```

--------------------------------

### Add BPMN Specs for Diffing

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Use these commands to add BPMN and DMN files to the diff utility. Specify the product name, BPMN file, and DMN files.

```console
./runner.py -e spiff_example.spiff.diffs add -p order_product \
    -b bpmn/tutorial/task_types.bpmn \
    -d bpmn/tutorial/product_prices.dmn
```

```console
./runner.py -e spiff_example.spiff.diffs add -p order_product \
    -b bpmn/tutorial/gateway_types.bpmn \
    -d bpmn/tutorial/{product_prices,shipping_costs}.dmn
```

```console
./runner.py -e spiff_example.spiff.diffs add -p order_product \
    -b bpmn/tutorial/{top_level,call_activity}.bpmn \
    -d bpmn/tutorial/{shipping_costs,product_prices}.dmn
```

```console
./runner.py -e spiff_example.spiff.diffs add -p order_product \
    -b bpmn/tutorial/{top_level_script,call_activity_script}.bpmn \
    -d bpmn/tutorial/shipping_costs.dmn
```

--------------------------------

### Run BPMN Process with Different Script Engines

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

These commands demonstrate how to run a BPMN process using different script environments. The first command runs with a regular environment, potentially allowing dangerous operations. The second command runs with a restricted environment, which will likely cause an error if imports are restricted.

```bash
./runner.py -e spiff_example.spiff.file add -p end_it_all -b bpmn/tutorial/dangerous.bpmn
./runner.py -e spiff_example.spiff.file
```

```bash
./runner.py -e spiff_example.spiff.restricted
```

--------------------------------

### Initialize File Serializer with Registry

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/serialization.rst.txt

Initialize a FileSerializer with a custom registry that includes methods for serializing custom objects. This serializer will write workflow data to disk.

```python
from SpiffWorkflow.spiff.serializer.config import SPIFF_CONFIG
from ..serializer.file import FileSerializer

registry = FileSerializer.configure(SPIFF_CONFIG)
registry.register(ProductInfo, product_info_to_dict, product_info_from_dict)
serializer = FileSerializer(dirname, registry=registry)
```

--------------------------------

### Import Serializer Configuration and Converters

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import DEFAULT_CONFIG and various converter classes for customizing BPMN serialization, including task specs and event definitions.

```python
from SpiffWorkflow.bpmn.serializer import DEFAULT_CONFIG
from SpiffWorkflow.bpmn.serializer.default import <TaskSpecConverter>
from SpiffWorkflow.bpmn.serializer.helpers import (
    TaskSpecConverter,
    EventDefinitionConverter,
    BpmnDataSpecificationConverter,
)
```

--------------------------------

### Get Subprocess Tasks in BPMN Workflow

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/workflows.html

Retrieves the first product customization task and then accesses only the tasks within that specific subprocess. Ensure the workflow object is properly instantiated.

```python
task = workflow.get_next_task(spec_name='customize_product')
subprocess = workflow.get_subprocess(task)
subprocess_tasks = subprocess.get_tasks()
```

--------------------------------

### Import Custom Task and Event Definition Parsers

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import TaskParser and EventDefinitionParser for customized parsing of specific BPMN elements.

```python
from SpiffWorkflow.bpmn.parser import TaskParser, EventDefinitionParser
```

--------------------------------

### Render User Instructions for Tasks

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/workflows.html

This function renders custom instructions for end-users, incorporating workflow data into a Jinja template. Ensure 'instructionsForEndUser' is present in task extensions.

```python
from jinja2 import Template

def set_instructions(self, task):
    user_input = self.ui._states['user_input']
    user_input.instructions = f'{self.task.task_spec.bpmn_name}\n\n'
    text = self.task.task_spec.extensions.get('instructionsForEndUser')
    if text is not None:
        template = Template(text)
        user_input.instructions += template.render(self.task.data)
    user_input.instructions += '\n\n'

```

--------------------------------

### Get Workflow Differences

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/diffs.html

This Python function retrieves a workflow instance and its specification, then calculates the differences between them. It returns a WorkflowDiff object for the top-level workflow and a dictionary of WorkflowDiff objects for any subprocesses.

```python
def diff_workflow(self, wf_id, spec_id):
    wf = self.serializer.get_workflow(wf_id)
    spec, deps = self.serializer.get_workflow_spec(spec_id)
    return diff_workflow(self.serializer.registry, wf, spec, deps)

```

--------------------------------

### Check Migration Safety

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Determines if a workflow is safe to migrate based on the state of its tasks and subprocesses. It checks if tasks that have been changed or removed have already completed or started, or if subprocesses have changed.

```python
def can_migrate(self, wf_diff, sp_diffs):

    def safe(result):
        mask = TaskState.COMPLETED|TaskState.STARTED
        tasks = result.changed + result.removed
        return len(filter_tasks(tasks, state=mask)) == 0

    for diff in sp_diffs.values():
        if diff is None or not safe(diff):
            return False
    return safe(wf_diff)
```

--------------------------------

### Manual Task Instructions Template

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

A Jinja template for displaying an order summary, including product name, quantity, and calculated order total.

```jinja
Order Summary
{{ product_name }}
Quantity: {{ product_quantity }}
Order Total: {{ order_total }}
```

--------------------------------

### Import BPMN Workflow and Event

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import the base BpmnWorkflow and BpmnEvent classes for creating and managing BPMN workflows.

```python
from SpiffWorkflow.bpmn import BpmnWorkflow, BpmnEvent
from SpiffWorfkflow import TaskState
```

--------------------------------

### Define Custom Start Event Spec

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/custom_task_spec.rst.txt

Create a custom task spec class that inherits from StartEventMixin and SpiffBpmnTask. This spec handles TimerEventDefinition by replacing it with NoneEventDefinition for custom timer management.

```python
from SpiffWorkflow.bpmn.specs.event_definitions import NoneEventDefinition
from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition
from SpiffWorkflow.bpmn.specs.mixins import StartEventMixin
from SpiffWorkflow.spiff.specs import SpiffBpmnTask

class CustomStartEvent(StartEventMixin, SpiffBpmnTask):

    def __init__(self, wf_spec, bpmn_id, event_definition, **kwargs):

        if isinstance(event_definition, TimerEventDefinition):
            super().__init__(wf_spec, bpmn_id, NoneEventDefinition(), **kwargs)
            self.timer_event = event_definition
        else:
            super().__init__(wf_spec, bpmn_id, event_definition, **kwargs)
            self.timer_event = None
```

--------------------------------

### Run Curses Application with JSON Serialization

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/application.html

Launch the curses-based user interface for running and examining workflows. This command assumes JSON file serialization is configured.

```bash
./runner.py -e spiff_example.spiff.file

```

--------------------------------

### List Workflow Instances

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Lists all saved instances of a workflow. Useful for identifying instances to diff or migrate.

```console
./runner.py -e spiff_example.spiff.diffs list_instances
```

--------------------------------

### Get Task Display Information

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/workflows.html

Retrieves display information for a given task, including its depth, state, name (prioritizing BPMN name), and lane. This is useful for UI representations of workflow tasks.

```python
def get_task_display_info(self, task):
    return {
        'depth': task.depth,
        'state': TaskState.get_name(task.state),
        'name': task.task_spec.bpmn_name or task.task_spec.name,
        'lane': task.task_spec.lane,
    }
```

--------------------------------

### Compare Specs with Dependencies

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Compare BPMN specifications including their dependencies. Use the '-d' flag to include dependent workflows in the diff output.

```console
./runner.py -e spiff_example.spiff.diffs diff_spec -d \
    -o 09400c6b-5e42-499d-964a-1e9fe9673e51 \
    -n f679a7ca-298a-4bff-8b2f-6101948715a9
```

--------------------------------

### Check Workflow Migration Safety

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/diffs.html

This Python function determines if a workflow is safe to migrate based on the states of its tasks and subprocesses. It checks if changed or removed tasks have been completed or started, or if subprocesses have changed.

```python
def can_migrate(self, wf_diff, sp_diffs):

    def safe(result):
        mask = TaskState.COMPLETED|TaskState.STARTED
        tasks = result.changed + result.removed
        return len(filter_tasks(tasks, state=mask)) == 0

    for diff in sp_diffs.values():
        if diff is None or not safe(diff):
            return False
    return safe(wf_diff)

```

--------------------------------

### Workflow Path Visualization

Source: https://spiffworkflow.readthedocs.io/en/latest/concepts.html

Illustrates a potential execution path of a workflow with a loop, showing how a specification can lead to multiple task instances.

```text
Start --> Select and Customize Product -> Continue Shopping? -> |
      /-------------------------------------------------------- /
      |
      |-> Select and Customize Product -> Continue Shopping? -> |
      /-------------------------------------------------------- /
      |
      |-> Select and Customize Product -> Continue Shopping?
```

--------------------------------

### Service Task BPMN XML Configuration

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

Example of a Service Task in BPMN XML, defining the operation name, parameters, and a post-script for processing the result. The `resultVariable` attribute stores the operation's output.

```xml
<bpmn:serviceTask id="Activity_1ln3xkw" name="Lookup Product Info">
  <bpmn:extensionElements>
    <spiffworkflow:serviceTaskOperator id="lookup_product_info" resultVariable="product_info">
      <spiffworkflow:parameters>
        <spiffworkflow:parameter id="product_name" type="str" value="product_name"/>
      </spiffworkflow:parameters>
    </spiffworkflow:serviceTaskOperator>
    <spiffworkflow:postScript>product_info = product_info_from_dict(product_info)</spiffworkflow:postScript>
  </bpmn:extensionElements>
  <bpmn:incoming>Flow_104dmrv</bpmn:incoming>
  <bpmn:outgoing>Flow_06k811b</bpmn:outgoing>
</bpmn:serviceTask>

```

--------------------------------

### Initialize SpiffBpmnParser

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/application.html

Initializes the built-in parser that handles both DMN files and Spiff BPMN extensions. No further customization is typically needed for this parser.

```python
parser = SpiffBpmnParser()
```

--------------------------------

### Instantiate a BPMN Workflow

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/workflows.rst.txt

Creates a new BPMN workflow instance from a given specification ID using the engine's serializer.

```python
def start_workflow(self, spec_id):
        spec, sp_specs = self.serializer.get_workflow_spec(spec_id)
        wf = BpmnWorkflow(spec, sp_specs, script_engine=self._script_engine)
        wf_id = self.serializer.create_workflow(wf, spec_id)
        return Instance(wf_id, workflow)
```

--------------------------------

### Migrate Workflow Instance via CLI

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/diffs.html

Command to initiate the migration of a workflow instance to a new specification. Requires the specification ID and the workflow instance ID.

```bash
./runner.py -e spiff_example.spiff.diffs migrate \
    -s 9da66c67-863f-4b88-96f0-76e76febccd0 \
    -w 4af0e043-6fd6-448d-85eb-d4e86067433e

```

--------------------------------

### Import Spiff Workflow Components

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import SpiffWorkflow components for parsing, specs, and serialization, tailored for the Spiff engine.

```python
from SpiffWorkflow.spiff.parser import SpiffBpmnParser, VALIDATOR
from SpiffWorkflow.spiff.specs import <TaskSpec>
from SpiffWorkflow.spiff.serializer import DEFAULT_CONFIG
```

--------------------------------

### Load BPMN Diagrams with Custom Script Engine

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/script_engine.html

This command loads BPMN diagrams that utilize a custom script engine, specifically one that includes custom objects like product information and shipping cost lookups. This allows the process to call these custom functions defined in Python.

```bash
./runner.py -e spiff_example.spiff.custom_object add -p order_product \
    -b bpmn/tutorial/{top_level_script,call_activity_script}.bpmn
```

--------------------------------

### Run Process with Restricted Engine

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/script_engine.rst.txt

Execute a BPMN process using a restricted script engine configuration. This demonstrates how to load a specific engine profile.

```console
./runner.py -e spiff_example.spiff.restricted
```

--------------------------------

### Diff Workflow Instances

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/diffs.html

Compare a specific workflow instance against a reference specification to identify changes in tasks and their states. Requires instance and specification IDs.

```bash
./runner.py -e spiff_example.spiff.diffs diff_workflow \
    -s 9da66c67-863f-4b88-96f0-76e76febccd0 \
    -w 4af0e043-6fd6-448d-85eb-d4e86067433e

```

--------------------------------

### Import Camunda Workflow Components

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import Camunda-specific components for parsing, specs, and serialization, enabling integration with Camunda BPMN models.

```python
from SpiffWorkflow.camunda.parser import CamundaParser
from SpiffWorkflow.camunda.specs import <TaskSpec>
from SpiffWorkfllw.camunda.serializer import DEFAULT_CONFIG
```

--------------------------------

### Migrate Workflow Command

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/diffs.rst.txt

Executes the migration of a workflow instance to a new specification using the command-line runner.

```console
./runner.py -e spiff_example.spiff.diffs migrate \
    -s 9da66c67-863f-4b88-96f0-76e76febccd0 \
    -w 4af0e043-6fd6-448d-85eb-d4e86067433e
```

--------------------------------

### Import DMN Parser, Mixin, and Serializer

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import components for integrating DMN (Decision Model and Notation) with BPMN workflows, including parsing, business rule task mixins, and serialization.

```python
from SpiffWorkflow.dmn.parser import BpmnDmnParser
from SpiffWorkflow.dmn.specs import BusinessRuleTaskMixin
from SpiffWorkflow.dmn.serializer import BaseBusinessRuleTaskConverter
```

--------------------------------

### Import Basic BPMN Parser and Validator

Source: https://spiffworkflow.readthedocs.io/en/latest/_sources/bpmn/imports.rst.txt

Import the BpmnParser and BpmnValidator for basic parsing and validation of BPMN models.

```python
from SpiffWorkflow.bpmn.parser import BpmnParser, BpmnValidator
```

--------------------------------

### Configure Task Logger

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/logging.html

Sets up the 'spiff.task' logger to output messages with a custom format including workflow and task specific information. Use this to monitor task state transitions.

```python
task_logger = logging.getLogger('spiff.task')
task_handler = logging.StreamHandler()
task_handler.setFormatter(logging.Formatter('%(asctime)s [%(name)s:%(levelname)s] (%(workflow_spec)s:%(task_spec)s) %(message)s'))
task_logger.addHandler(task_handler)
```

--------------------------------

### Add Workflow Spec via Command Line

Source: https://spiffworkflow.readthedocs.io/en/latest/bpmn/application.html

Use this command to add a workflow specification to the application, specifying BPMN and DMN files. This utilizes shell completion for file paths.

```bash
./runner.py -e spiff_example.spiff.file add \
   -p order_product \
   -b bpmn/tutorial/{top_level,call_activity}.bpmn \
   -d bpmn/tutorial/{product_prices,shipping_costs}.dmn

```
Tokens:
Raw