orcher-sdk 0.4.12

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.

A workflow is marked the same way: an exception whose non_retryable attribute is true, raised from workflow code, fails the workflow without a retry, whatever its retry policy allows. Any other exception a workflow raises is retried as its policy says.

A workflow that catches the failure reads its type and whether the task was cancelled from the TaskError:

try:
    await ctx.execute_task(charge, order_id)
except TaskError as e:
    if e.is_cancelled:
        ...  # the task was cancelled
    elif e.error_type == "CardDeclined":
        ...  # ask for another card

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.

A worker runs at most max_concurrent_workflows workflow activations, max_concurrent_tasks tasks and max_concurrent_actor_operations actor operations at once (set on the builder), and takes no more work from the engine while it is full. It registers with the engine when it starts, and the engine sees which SDK and release it runs.

await worker.shutdown() stops it taking work at once, and gives the work it is running the shutdown grace (shutdown_grace_time, 30 seconds unless set) to finish. Workflow activations still unfinished after that are handed back to the engine for another worker; tasks still running are asked to cancel (ctx.cancellation_token) and are retried once they time out. shutdown(force=True) cancels the running work without waiting for it.