Workflows

class chancy.plugins.workflow.WorkflowPlugin(*, polling_interval=30, max_workflows_per_run=1000, pruning_rule=None, max_pending_workflows=10000)

Bases: Plugin

Support for dependency-based workflows.

Workflows allow you to easily model complex processes that involve multiple jobs, each of which may depend on the completion of one or more other jobs. Workflows are modeled as a directed acyclic graph (DAG) where each node represents a job and each edge represents a dependency between two jobs.

Workflows are implemented on top of the existing Chancy job system, meaning you can use all the existing job features, such as job retries, timeouts, scheduling, and so on.

Note

This plugin is enabled by default, you only need to provide it in the list of plugins to customize its arguments or if no_default_plugins is set to True.

Enable the plugin by adding it to the list of plugins in the Chancy constructor:

from chancy.plugins.leadership import Leadership
from chancy.plugins.workflow import WorkflowPlugin

async with Chancy(
    "postgresql://localhost/postgres",
    plugins=[WorkflowPlugin()],
) as chancy:
    ...

Example

We’ll create a simple workflow runs the “top” job first, then the “left” and “right” jobs in parallel, and finally the “bottom” job:

example_workflow.py

 import asyncio
 from chancy import Chancy, job
 from chancy.plugins.leadership import Leadership
 from chancy.plugins.workflow import Workflow, WorkflowPlugin

 @job()
 async def top():
     print(f"Top")

 @job()
 async def left():
     print(f"Left")

 @job()
 async def right():
     print(f"Right")

 @job()
 async def bottom():
     print(f"Bottom")

 async def main():
     async with Chancy(
         "postgresql://localhost/postgres",
         plugins=[Leadership(), WorkflowPlugin()]
     ) as chancy:
         workflow = (
             Workflow("example")
             .add("top", top)
             .add("left", left, ["top"])
             .add("right", right, ["top"])
             .add("bottom", bottom, ["left", "right"])
         )
         await WorkflowPlugin.push(chancy, workflow)

 if __name__ == "__main__":
     asyncio.run(main())

If we visualize our newly created workflow using generate_dot(), we get:

The full Workflow API is a little verbose if you just want to run a series of jobs in a specific order. In that case, you can use the Sequence class to create a workflow from a list of jobs.

Django Integration

This plugin can be made available to the Django ORM and Admin interface.

To enable this, you need to add the following to your Django settings:

INSTALLED_APPS = [
    ...,
    "chancy.plugins.workflow.django",
]

You can then query the workflows and steps using the Django ORM:

from chancy.plugins.workflow.django.models import (
    Workflow,
    WorkflowStep
)

workflow = Workflow.objects.get(id="...")

completed_steps = WorkflowStep.objects.filter(
    workflow=workflow,
    state=QueuedJob.State.SUCCEEDED
)

param polling_interval:

The interval between recovery polls. Notification traffic does not reset this interval.

param max_workflows_per_run:

The maximum number of workflows to process in one polling or notification batch. Recovery polls continue where the previous batch left off.

param max_pending_workflows:

The maximum number of distinct workflows awaiting event-driven processing. Further notifications rely on periodic polling.

class Rules

Bases: object

class Age

Bases: Rule

to_sql()

Return type:

Composable

api_plugin()

If this plugin has an associated API component, returns the import string for the plugin.

Return type:

str | None

async cleanup(chancy)

Clean up any resources used by the plugin.

Should return either None, if no work was done, or the number of rows cleaned up.

Note

Normally, you don’t need to call this yourself. The Pruner plugin will call the cleanup method of all other registered plugins.

Parameters:

chancy (Chancy)

Return type:

int | None

async classmethod fetch_workflow(chancy, id_)

Fetch a single workflow from the database.

Parameters:

• chancy (Chancy) – The Chancy application.

• id – The ID of the workflow to fetch.

• id_ (str)

Returns:

The workflow, or None if it does not exist.

Return type:

Workflow | None

async classmethod fetch_workflow_ex(cursor, chancy, id_)

Fetch a single workflow from the database.

This method is a lower-level version of fetch_workflow that accepts an existing cursor object, allowing it to be used in transactions.

Parameters:

• cursor (AsyncCursor[dict[str, Any]]) – The cursor to use for the query.

• chancy (Chancy) – The Chancy application.

• id – The ID of the workflow to fetch.

• id_ (str)

Returns:

The workflow, or None if it does not exist.

Return type:

Workflow | None

async classmethod fetch_workflows(chancy, *, states=None, ids=None, limit=100)

Fetch workflows from the database, optionally matching the given conditions.

Parameters:

• chancy (Chancy) – The Chancy application.

• states (list[str] | None) – A list of states to match.

• ids (list[str] | None) – A list of IDs to match.

• limit (int) – The maximum number of workflows to fetch.

Returns:

A list of workflows.

Return type:

list[Workflow]

async static fetch_workflows_ex(cursor, chancy, *, states=None, ids=None, limit=100)

Fetch workflows from the database, optionally matching the given conditions.

This method is a lower-level version of fetch_workflows that accepts an existing cursor object, allowing it to be used in transactions.

Parameters:

• cursor (AsyncCursor[dict[str, Any]]) – The cursor to use for the query.

• chancy (Chancy) – The Chancy application.

• states (list[str] | None) – A list of states to match.

• ids (list[str] | None) – A list of IDs to match.

• limit (int) – The maximum number of workflows to fetch.

Returns:

A list of workflows.

Return type:

list[Workflow]

static generate_dot(workflow, output)

Generate a DOT file representation of the workflow.

Parameters:

• workflow (Workflow) – The Workflow object to visualize.

• output (TextIO) – A file-like object to write the DOT content to.

static get_dependencies()

Get the identifiers of all plugins this plugin depends on, if any.

Plugins that depend on the presence of other plugins will refuse to start if those dependencies are not met.

Return type:

list[str]

static get_identifier()

Returns a unique identifier for this plugin.

This identifier should be unique across all active plugins. If a custom plugin provides compatible functionality to a built-in plugin, it may use the same identifier as the built-in plugin.

Return type:

str

get_tables()

Get the names of all tables this plugin is responsible for.

Return type:

list[str]

migrate_key()

Get the migration key for this plugin, if it has any.

Return type:

str

migrate_package()

Get the package name that contains the migration scripts for this plugin, if it has any.

Return type:

str

async on_jobs_updated_in_transaction(*, worker, jobs, cursor)

Called after a batch’s SQL updates, before its transaction commits.

Use the supplied cursor for database work that must commit atomically with the job updates. Raising an exception rolls back the batch. This hook may run again on retry; avoid external or in-memory side effects. Use on_job_updated() for work that requires committed updates.

Parameters:

• worker (Worker) – The worker saving the updates.

• jobs (Sequence[QueuedJob]) – Only updates whose claims matched and whose database rows were updated, in their original order, including repeated accepted updates to the same job. Rejected updates are omitted; this hook is not called if none were accepted.

• cursor (AsyncCursor[dict[str, Any]]) – The cursor for the current update transaction.

async poll(worker, chancy, cursor, *, changes=None)

Process the next batch of up to max_workflows_per_run pending or running workflows.

Workflows are taken in ID order within a sweep whose upper bound is fixed at its start. Newer workflows join the next sweep, so arrivals cannot indefinitely postpone revisiting older workflows. A batch stops at the end of its sweep even if it has room for more workflows.

Parameters:

• worker (Worker) – The worker processing the workflows.

• chancy (Chancy) – The Chancy application.

• cursor (AsyncCursor[dict[str, Any]]) – The cursor to use for the query.

• changes (list | None)

Returns:

The number of workflows processed.

Return type:

int

async process_pending(worker, chancy, cursor, *, changes=None)

Process one batch of notified workflows using the caller’s transaction.

Duplicate notifications are coalesced until their batch starts. Locked workflows and dropped hints are recovered by periodic polling.

Returns:

The number of workflows processed.

Parameters:

• worker (Worker)

• chancy (Chancy)

• cursor (AsyncCursor[dict[str, Any]])

• changes (list | None)

Return type:

int

async static process_workflow(cursor, chancy, workflow, worker)

Process a single iteration of the given workflow, progressing the state of each step and the overall workflow as necessary.

Parameters:

• cursor (AsyncCursor) – The cursor to use for the query.

• chancy (Chancy) – The Chancy application.

• workflow (Workflow) – The workflow to process.

• worker (Worker) – The worker processing the workflow.

Returns:

True if the workflow was updated, False otherwise.

Return type:

bool

async classmethod push(chancy, workflow)

Push new workflow to the database.

If the workflow already exists in the database, it will be updated instead.

Parameters:

• chancy (Chancy) – The Chancy application.

• workflow (Workflow) – The workflow to push.

Returns:

The UUID of the newly created workflow.

Return type:

str

async classmethod push_ex(cursor, chancy, workflow, worker=None)

Push new workflow to the database.

This method is a lower-level version of push that accepts an existing cursor object, allowing it to be used in transactions.

If the workflow already exists in the database, it will be updated instead.

Parameters:

• cursor (AsyncCursor[dict[str, Any]]) – The cursor to use for the query.

• chancy (Chancy) – The Chancy application.

• workflow (Workflow) – The workflow to push.

• worker (Worker) – Associated worker. The scheduler emits metrics after commit.

Returns:

The UUID of the newly created workflow.

Raises:

• EmptyWorkflowError – If the workflow has no steps.

• CircularDependencyError – If a circular dependency is detected.

• InvalidDependencyError – If a dependency references a non-existent step.

Return type:

str

async run(worker, chancy)

Runs the plugin.

This function can and should run indefinitely, as it will be cancelled when the worker is stopped.

Parameters:

• worker (Worker)

• chancy (Chancy)

async classmethod wait_for_workflow(chancy, workflow_id, *, interval=1, timeout=None)

Wait for a workflow to complete.

This method will loop until the workflow referenced by the provided ID has completed. The interval parameter controls how often the workflow status is checked. This will not block the event loop, so other tasks can run while waiting for the workflow to complete.

Example

workflow = Workflow("example")
workflow.add("step1", job1)
workflow.add("step2", job2, ["step1"])

workflow_id = await WorkflowPlugin.push(chancy, workflow)
completed_workflow = await WorkflowPlugin.wait_for_workflow(
    chancy,
    workflow_id,
    timeout=300  # 5 minute timeout
)

param chancy:

The Chancy application.

param workflow_id:

The ID of the workflow to wait for.

param interval:

The number of seconds to wait between checks.

param timeout:

The maximum number of seconds to wait for the workflow to complete. If not provided, the method will wait indefinitely.

raises asyncio.TimeoutError:

If the timeout is reached before the workflow completes.

raises KeyError:

If the workflow does not exist.

return:

The completed Workflow object.

Parameters:

• chancy (Chancy)

• workflow_id (str)

• interval (int)

• timeout (float | None)

Return type:

Workflow

Parameters:

• polling_interval (int)

• max_workflows_per_run (int)

• pruning_rule (SQLAble | None)

• max_pending_workflows (int)

class chancy.plugins.workflow.Workflow(name, steps=<factory>, state=State.PENDING, id=<factory>, created_at=None, updated_at=None)

Bases: object

A collection of jobs with dependencies between steps.

Define steps using add() or add_group(), then submit the workflow using WorkflowPlugin.push(). Each step is queued once all of its dependencies have succeeded. Steps without dependencies can run in parallel.

For jobs that should run one after another, use Sequence.

Parameters:

• name (str)

• steps (dict[str, WorkflowStep])

• state (State)

• id (str)

• created_at (datetime | None)

• updated_at (datetime | None)

class State(*values)

Bases: Enum

COMPLETED = 'completed'

FAILED = 'failed'

PENDING = 'pending'

RUNNING = 'running'

add(step_id, job, dependencies=None)

Add a step to the workflow.

workflow.add("step_1", job)
workflow.add("step_2", job, ["step_1"])

Parameters:

• step_id (str) – The ID of the step.

• job (Job | IsAJob) – The job to execute.

• dependencies (list[str] | None) – A list of step IDs that this step depends on.

Return type:

Workflow

add_group(jobs, dependencies=None)

Add a group of steps to the workflow.

This is a convenience method for adding multiple steps to the workflow at once that are all dependent on the same set of dependencies.

workflow = Workflow("my_workflow")
workflow.add("setup", setup_job)
workflow.add_group([
    ("step_1", job_1),
    ("step_2", job_2),
    ("step_3", job_3),
], ["setup"])

Parameters:

• jobs (list[tuple[str, Job | IsAJob]]) – A list of tuples of (step_id, job).

• dependencies (list[str] | None) – A list of step IDs that this step depends on.

Return type:

Workflow

created_at: datetime | None = None

The time the workflow was created.

id: str

The unique ID of a specific run of the workflow.

property is_complete: bool

property is_running: bool

name: str

state: State = 'pending'

The current state of the workflow.

steps: dict[str, WorkflowStep]

A dictionary of steps in the workflow, keyed by step ID.

property steps_by_state: dict[State, list[WorkflowStep]]

updated_at: datetime | None = None

The time the workflow was last updated.

validate()

Validate the workflow’s dependency graph.

This method checks for: - Empty workflows (no steps) - Circular dependencies (cycles in the dependency graph) - Invalid dependencies (references to non-existent steps)

Raises:

• EmptyWorkflowError – If the workflow has no steps.

• CircularDependencyError – If a circular dependency is detected.

• InvalidDependencyError – If a dependency references a non-existent step.

Return type:

None

class chancy.plugins.workflow.WorkflowStep(job: chancy.job.Job | chancy.job.IsAJob, step_id: str, dependencies: list[str] = <factory>, state: chancy.job.QueuedJob.State | None = <State.PENDING: 'pending'>, job_id: str | None = None)

Bases: object

Parameters:

• job (Job | IsAJob)

• step_id (str)

• dependencies (list[str])

• state (State | None)

• job_id (str | None)

dependencies: list[str]

A list of step IDs that this step depends on.

job: Job | IsAJob

The job to execute when this step is ready.

job_id: str | None = None

The unique ID of a running Job which is associated with this step.

state: State | None = 'pending'

The current state of the step.

step_id: str

The unique ID of the step.

class chancy.plugins.workflow.Sequence(name, jobs=None)

Bases: object

A sequential workflow.

Sequences are a special case of workflows, where each step depends on the previous step. This forms a linear chain of jobs that are executed in order.

Sequences are useful for defining sequences of jobs that must be executed in order, without the complexity of full workflows.

Example

example_sequence.py

 import asyncio
 from chancy import Chancy, job
 from chancy.plugins.workflow import Sequence

 @job()
 async def first():
     print("First")

 @job()
 async def second():
     print("Second")

 @job()
 async def third():
     print("Third")

 async def main():
     async with Chancy("postgresql://localhost/postgres") as chancy:
         sequence = Sequence("example_workflow", [first, second, third])
         await sequence.push(chancy)

 if __name__ == "__main__":
     asyncio.run(main())

add(job)

Add a job to the sequence.

workflow = (
    Sequence("example_sequence")
    .add(first)
    .add(second)
    .add(third)
)

Parameters:

job (Job | IsAJob) – The job to add.

Return type:

Self

async push(chancy)

Push a sequence to the database.

Parameters:

chancy (Chancy) – The Chancy application.

Returns:

The UUID of the newly created chain.

Return type:

str

Parameters:

• name (str)

• jobs (list[Job | IsAJob] | None)

class chancy.plugins.workflow.api.WorkflowApiPlugin(api)

Bases: ApiPlugin

API plugin for workflows.

async static get_workflow(request, *, chancy, worker)

Get a single workflow by ID.

async static get_workflows(request, *, chancy, worker)

List workflows in descending ID order with optional filtering.

Pass the last returned ID as before to fetch the next page of lower IDs with the same filters. Creation and update times do not affect this order. Pages are live results, not a snapshot: workflows may be removed or stop matching filters between requests.

Set pagination=true to receive an object with items, has_more, and next_cursor (an ID to pass as before, or null on the last page). Otherwise the response remains an array. limit must be positive and is capped at 100.

total_steps counts every workflow step. waiting_steps counts steps not yet submitted (no job ID), including dependency-blocked steps. pending_steps counts jobs already queued in pending state. A step whose job was deleted still contributes to total_steps but is not counted as waiting or in any job state.

name()

Get the name of the plugin.

routes()

Get a list of routes to add to the API.

exception chancy.plugins.workflow.EmptyWorkflowError

Raised when a workflow has no steps.

exception chancy.plugins.workflow.CircularDependencyError

Raised when a circular dependency is detected in a workflow.

exception chancy.plugins.workflow.InvalidDependencyError

Raised when a dependency references a non-existent step.