Workflows¶

Experimental feature

This feature is experimental and should be considered unstable and unsupported. API may change at any time and without backwards compatibility.

A workflow chains tasks into a multi-step process where each step can use the results of the previous ones. Write it as a generator function: yield tasks, get their finished versions back.

Defining a workflow¶

from sheppy import task
from sheppy.experimental import workflow

@task
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow
def greetings_workflow(names: list[str]):
    t1 = yield say_hello("Alice")
    t2 = yield say_hello("Bob")
    tx = yield [say_hello(name) for name in names]  # fan-out: steps run in parallel

    return "\n".join([t1.result, t2.result] + [t.result for t in tx])

Yield a single Task for a sequential step, or a list[Task] to run steps in parallel.
The value of the yield expression is the finished Task: read task.result for the return value and check task.exception for failures.
The generator's return value becomes the workflow's final_result.
Calling the decorated function returns a Workflow instance; nothing runs yet.

The generator must be a plain (non-async) generator function, and it must be deterministic. On each resume it is replayed from the start with finished tasks fed back in, so the sequence of yields may only depend on previous task results, never on external state or randomness. Depends and CURRENT_TASK are not allowed in workflow functions.

Running a workflow¶

add_workflow() starts the run and wait_for_workflow() blocks until it completes or fails:

result = await queue.experimental.add_workflow(greetings_workflow(["Alex", "John"]))
wf = await queue.experimental.wait_for_workflow(result.workflow.id)
print(wf.final_result)

examples/workflows/simple.py is self-contained (in-memory backend plus an in-process worker):

examples/workflows/simple.py

import asyncio

from sheppy import InMemoryBackend, Queue, Worker, task
from sheppy.experimental import workflow


@task
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"


@workflow
def greetings_workflow(names: list[str]):
    t1 = yield say_hello("Alice")
    t2 = yield say_hello("Bob")
    tx = yield [say_hello(name) for name in names]  # fan-out style

    return "\n".join([t1.result, t2.result] + [t.result for t in tx])


queue = Queue(InMemoryBackend(instant_processing=False))


async def main():
    # start worker in background
    worker = Worker("default", backend=queue.backend)
    worker_process = asyncio.create_task(worker.work())

    result = await queue.experimental.add_workflow(greetings_workflow(["Alex", "John"]))

    # wait for the workflow to finish
    wf = await queue.experimental.wait_for_workflow(result.workflow.id)

    # stop worker
    worker_process.cancel()

    if wf is None or wf.exception:
        print(f"Workflow failed: {wf.exception if wf else 'not found'}")
    else:
        print(wf.final_result)


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

$ python examples/workflows/simple.py
Hello, Alice!
Hello, Bob!
Hello, Alex!
Hello, John!

Handling failures¶

A failed task does not fail the workflow by itself. Check task.exception on the finished task and take action (roll-back, notify, retry), or raise to fail the whole workflow, which sets wf.exception. examples/workflows/fan_in_fan_out.py shows the pattern:

examples/workflows/fan_in_fan_out.py

import asyncio

from sheppy import InMemoryBackend, Queue, Worker, task
from sheppy.experimental import workflow

ADMIN_EMAILS = [
    "[email protected]",
    "[email protected]",
    "[email protected]",
]


@task
async def cleanup_old_data(days: int = 7):
    if days > 7:  # deterministic "random" failure
        raise Exception("some random failure happened")

    return "everything ok!"


@task
async def some_cleanup_at_the_end():
    return True


@task
async def rollback_changes():
    return True


@task
async def send_notification(to: str, subject: str):
    print(f"Sending email to {to}, subject {subject}")


@workflow
def daily_cleanup(days_to_clean: int):
    result_task = yield cleanup_old_data(days=days_to_clean)

    if result_task.exception:
        yield rollback_changes()
        yield [send_notification(email, "Oh no, daily cleanup failed!") for email in ADMIN_EMAILS]

        raise Exception("Cleanup failed, notifications were sent")  # fail the workflow

    yield some_cleanup_at_the_end()
    yield send_notification("[email protected]", "Cleanup finished")

    return "Daily cleanup finished successfully"


queue = Queue(InMemoryBackend(instant_processing=False))


async def main():
    # start worker in background
    worker = Worker("default", backend=queue.backend)
    worker_process = asyncio.create_task(worker.work())

    result = await queue.experimental.add_workflow(daily_cleanup(30))

    # wait for the workflow to finish (successfully or not)
    wf = await queue.experimental.wait_for_workflow(result.workflow.id)

    # stop worker
    worker_process.cancel()

    if wf is None:
        print("Workflow not found")
    elif wf.exception:
        print(f"Workflow failed as expected: {wf.exception}")
    else:
        print(f"Workflow completed: {wf.final_result}")


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

Running it makes cleanup_old_data raise; the workflow rolls back, fans out three notifications, and fails as designed:

$ python examples/workflows/fan_in_fan_out.py
Task failed function=examples.workflows.fan_in_fan_out:cleanup_old_data queue=default task_id=0adf46fa-... error=Exception: some random failure happened
Workflow failed queue=default workflow_id=b1e4879b-... error=Exception: Cleanup failed, notifications were sent
Sending email to [email protected], subject Oh no, daily cleanup failed!
Sending email to [email protected], subject Oh no, daily cleanup failed!
Sending email to [email protected], subject Oh no, daily cleanup failed!
Workflow failed as expected: Exception: Cleanup failed, notifications were sent

Notes and limits¶

Workflow state is persisted in the backend, so workflows survive restarts, but the workflow function must stay importable at its original module:function path.
Nested workflows: yield a Workflow from another @workflow function and it runs inline within the parent. The yield expression receives the nested generator's return value.
Workflow steps inherit the workflow's resolved ttl and error_ttl defaults. Nested workflows inherit from their parent.
In tests, TestQueue.experimental.process_workflow() drives a workflow synchronously to completion.