Activities

Activities are where non-deterministic work happens: database calls, HTTP requests, file I/O, and anything else that touches the outside world.

Defining activities

Use @activity.defn() for standalone functions:

from cadence import activity

@activity.defn()
async def fetch_order(order_id: str) -> dict:
    # HTTP call, DB query, etc.
    return {"id": order_id, "status": "pending"}

@activity.defn()
def send_email(to: str, subject: str) -> None:
    # synchronous activities work too
    ...

Use name= to give the activity an explicit name:

@activity.defn(name="fetch-order-v2")
async def fetch_order_v2(order_id: str) -> dict:
    ...

Method activities

Group related activities into a class using @activity.method():

from cadence import activity

class OrderActivities:
    @activity.method()
    async def fetch_order(self, order_id: str) -> dict:
        ...

    @activity.method()
    async def update_status(self, order_id: str, status: str) -> None:
        ...

Register an instance with the registry:

registry.register_activities(OrderActivities())

Registering activities

from cadence.worker import Registry

registry = Registry()

# Standalone function
registry.register_activity(fetch_order)

# All methods on an instance
registry.register_activities(OrderActivities())

Or use the registry decorator directly:

@registry.activity()
async def process_payment(amount: float) -> str:
    ...

Executing activities from a workflow

Inside a workflow, call execute_activity with the activity name, expected return type, arguments, and options:

from datetime import timedelta
from cadence.workflow import execute_activity

@registry.workflow()
class OrderWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        order = await execute_activity(
            "fetch_order",
            dict,
            order_id,
            start_to_close_timeout=timedelta(minutes=5),
        )
        await execute_activity(
            "send_email",
            type(None),
            order["email"],
            "Your order is ready",
            start_to_close_timeout=timedelta(seconds=30),
        )
        return "done"

You can also execute a decorated activity definition directly. with_options preserves its parameter and result types:

order = await fetch_order.with_options(
    start_to_close_timeout=timedelta(minutes=5),
).execute(order_id)

Activity options

The Python SDK starts with a 1-hour schedule_to_close_timeout and a 10-second schedule_to_start_timeout, then applies the options supplied by the workflow. If start_to_close_timeout or heartbeat_timeout is omitted, each is set to the effective Schedule-to-Close timeout. Set explicit values when these defaults do not fit the activity.

Option	Description
start_to_close_timeout	Max time for one activity attempt (default: effective Schedule-to-Close timeout)
schedule_to_close_timeout	Max total time including scheduling and all retries (default: 1 hour)
schedule_to_start_timeout	Max time waiting in the task list before execution starts (default: 10 seconds)
heartbeat_timeout	Max time between heartbeats for long-running activities (default: effective Schedule-to-Close timeout)
task_list	Override the task list for this activity
retry_policy	Retry policy (see Retries)

Heartbeating

Long-running activities should call activity.heartbeat() periodically so the server can detect failures and reschedule:

from cadence import activity
import asyncio

@activity.defn()
async def process_large_file(file_path: str) -> int:
    rows_processed = 0
    with open(file_path) as f:
        for line in f:
            process_line(line)
            rows_processed += 1
            if rows_processed % 1000 == 0:
                activity.heartbeat(rows_processed)  # report progress
            await asyncio.sleep(0)  # yield control
    return rows_processed

Pass progress details to heartbeat() and retrieve them on restart with activity.heartbeat_details().

Cancellation

Cancellation is delivered to activities through heartbeat responses. Long-running activities must set heartbeat_timeout and heartbeat regularly to receive cancellation promptly.

Async activities receive asyncio.CancelledError after a heartbeat observes the cancellation request:

import asyncio
from cadence import activity

@activity.defn()
async def cancellable_activity() -> None:
    try:
        while True:
            await do_work()
            activity.heartbeat()
    except asyncio.CancelledError:
        await clean_up()
        raise

Synchronous activities can use activity.is_cancelled(), activity.raise_if_cancelled(), or activity.wait_for_cancelled(timeout). Raise cadence.error.ActivityCancelledError after cleanup to report the activity as cancelled.

See Cancellation for complete workflow and activity examples.

Activity context

Inside an activity, use the activity module to access context:

@activity.defn()
async def my_activity() -> None:
    info = activity.info()
    print(info.activity_id)
    print(info.workflow_id)
    print(info.workflow_run_id)
    print(info.attempt)
    print(info.heartbeat_timeout)

activity.client() returns the worker's client from an async activity; it is not supported in synchronous activities. activity.in_activity() reports whether the current code is running in an activity context.