orcher-sdk 0.4.7

A timer is recorded by the engine, so no worker is busy while it runs, and a restart doesn't reset it:

from datetime import timedelta

from orcher import WorkflowContext, workflow


@workflow(name="trial")
async def trial(ctx: WorkflowContext, email: str) -> None:
    await ctx.sleep(timedelta(days=14))
    await ctx.execute_task(send_trial_ended, email=email)

A workflow can park until a named event arrives, optionally with a deadline:

from datetime import timedelta

from orcher import WorkflowContext, workflow


@workflow(name="approval")
async def approval(ctx: WorkflowContext, request_id: str) -> str:
    try:
        approved = await ctx.wait_for_event_with_timeout("approved", timedelta(days=3))
    except TimeoutError:
        return f"{request_id} expired"
    return f"{request_id} approved" if approved else f"{request_id} rejected"

Send the event from a client:

handle = await client.get_workflow("approval-42")
await handle.send_event("approved", True)

Cancel it the same way, optionally limiting how long its cleanup may take before the engine terminates it. Engines from before cancellation cleanup ignore the limit and cancel at once:

await handle.cancel(cleanup_timeout=timedelta(seconds=30))

from orcher import WorkflowContext, workflow


@workflow(name="ship-order")
async def ship_order(ctx: WorkflowContext, order_id: str) -> str:
    label = await ctx.execute_child_workflow("print-label", args=(order_id,))
    return f"{order_id} shipped with {label}"

ctx.start_child_workflow() returns a handle instead, to run several children at once and collect their results later.

A failure's type is the exception's class name (CardDeclined, not payments.CardDeclined). List the types never to retry in the task's policy. An exception whose non_retryable attribute is true is never retried, whatever the policy allows:

from orcher import RetryPolicy, TaskContext, task


class CardDeclined(Exception):
    pass


class AccountClosed(Exception):
    non_retryable = True


@task(
    name="charge",
    retry_policy=RetryPolicy(max_attempts=5, non_retryable_error_types=["CardDeclined"]),
)
async def charge(ctx: TaskContext, order_id: str) -> str:
    # Runs once:
    raise CardDeclined(f"card declined for {order_id}")
    # Also runs once, listed or not:
    # raise AccountClosed(f"account closed for {order_id}")

Any other exception is retried. non_retryable can also be set on a single instance before it is raised.

Workers and clients send and receive gRPC messages of up to 32 MiB; set ORCHER_MAX_MESSAGE_BYTES to change that. The engine accepts one payload (an input, a result, an event) of up to 8 MiB unless configured otherwise. A task whose result is too large fails straight away with a PayloadTooLarge failure that says how large it was, and is not retried: store large data elsewhere and pass a reference.

ctx.time reads the time the engine recorded, and ctx.random is seeded from the run, so both give the same answer every time the workflow is replayed:

from orcher import WorkflowContext, workflow


@workflow(name="invoice")
async def invoice(ctx: WorkflowContext, customer: str) -> str:
    number = ctx.random.uuid()
    issued_at = ctx.time.now()
    return f"invoice {number} for {customer}, issued at {issued_at.isoformat()}"

orcher.testing runs a workflow without an engine, with its tasks mocked by name. With pytest and pytest-asyncio:

import pytest

from orcher.testing import TestWorkflowEnvironment

from orders import confirm_order


@pytest.mark.asyncio
async def test_confirms_the_order() -> None:
    env = TestWorkflowEnvironment()
    env.mock_task("send-confirmation").returns("sent")

    receipt = await env.execute_workflow(
        confirm_order, {"order_id": "order-1", "email": "ada@example.com"}
    )

    assert receipt == "sent"
    env.assert_task_called_with(
        "send-confirmation", {"order_id": "order-1", "email": "ada@example.com"}
    )

Mocks can also raise, return a sequence or compute their result, and the environment controls the clock.