Decorators

The decorator API wraps individual functions in story or stage context managers. Decorators detect whether the function is a coroutine and apply async with or with accordingly.

@runtime_narrative_stage

Wraps a function in a stage(). The stage name defaults to the function name with underscores replaced by spaces and title-cased when called with no arguments.

python

from runtime_narrative import runtime_narrative_stage

@runtime_narrative_stage("Load CSV")           # explicit name
def load_csv() -> list[str]:
    return ["alice", "bob"]

@runtime_narrative_stage()                     # inferred name: "Validate Data"
def validate_data(rows: list[str]) -> None:
    if not rows:
        raise ValueError("empty dataset")

@runtime_narrative_stage("Insert Records")
async def insert_records(rows: list[str]) -> int:
    return len(rows)

@runtime_narrative_story

Wraps a function in a story(). Accepts all the same keyword arguments as story(). The story name defaults to the function name title-cased when called with no arguments.

python

from runtime_narrative import runtime_narrative_story, runtime_narrative_stage

@runtime_narrative_story(
    "Import Customers",
    failure_diagnostics="rich",
)
def import_customers() -> None:
    rows = load_csv()
    validate_data(rows)
    insert_records_sync(rows)

Function composition pattern

Decorate stages independently, then call them from inside a decorated story. The stage functions can live in different modules — the story context is found via ContextVar so no explicit passing is needed.

python

from runtime_narrative import runtime_narrative_story, runtime_narrative_stage

@runtime_narrative_stage("Load CSV")
def load_csv() -> list[str]:
    return ["alice", "bob"]

@runtime_narrative_stage("Validate Data")
def validate(rows: list[str]) -> None:
    assert rows, "empty dataset"

@runtime_narrative_stage("Insert Records")
def insert(rows: list[str]) -> int:
    return len(rows)

@runtime_narrative_story("Import Customers")
def run() -> None:
    rows = load_csv()
    validate(rows)
    count = insert(rows)
    print(f"Inserted {count} records")

try:
    run()
except Exception:
    pass

Async decorator composition

python

import asyncio
from runtime_narrative import runtime_narrative_story, runtime_narrative_stage

@runtime_narrative_stage("Fetch Users")
async def fetch_users() -> list[dict]:
    await asyncio.sleep(0.1)
    return [{"id": 1, "name": "Alice"}]

@runtime_narrative_stage("Sync to CRM")
async def sync_to_crm(users: list[dict]) -> None:
    await asyncio.sleep(0.2)

@runtime_narrative_story("CRM Sync")
async def run_sync() -> None:
    users = await fetch_users()
    await sync_to_crm(users)

asyncio.run(run_sync())

Next step

Continue to Auto-Instrumentation for wrapping whole classes and modules with zero call-site changes, or revisit stage() for the underlying context manager.