OOP Workflow Example

examples/oop_workflow.py shows how to keep workflow state on a normal Python object while still using decorator-based Astrum tasks.

Run it

python examples/oop_workflow.py

Expected output:

completed
A-001 in APAC: 129.60 USD

Pattern

Create a registry, decorate methods inside a class, instantiate the class, and pass the instance through AstrumConfig(class_instances=[...]).

workflow = SchedulerRegistry("oop_workflow")


class OrderService:
    def __init__(self, region: str) -> None:
        self.region = region
        self.tax_rate = 0.08

    @workflow.task("load_order")
    async def load_order(self) -> dict:
        return {"order_id": "A-001", "region": self.region, "subtotal": 120}

Downstream methods can use Ref/F exactly like module-level functions:

@workflow.task("price_order")
async def price_order(
    self,
    subtotal: Ref[int, F("load_order", "subtotal")],
) -> dict:
    total = round(subtotal * (1 + self.tax_rate), 2)
    return {"total": total, "currency": "USD"}

The scheduler injects self automatically for decorated unbound instance methods:

service = OrderService(region="APAC")
report = await workflow.run(
    target_tasks=["format_receipt"],
    config=AstrumConfig(
        class_instances=[service],
        skip_type_check=True,
        silence_warnings=True,
    ),
)

Manual DAGs are different: if you pass service.method directly to DynamicScheduler, Python has already bound self, so class_instances is not needed.