Simple Retry

"""
01_simple_retry.py: Task retries with a fixed delay.

Demonstrates:
- @task(retries=3, retry_delay_seconds=0.05)
- Task that fails on the first two attempts and succeeds on the third
"""

from dagy import flow, task

_attempt_counter: dict[str, int] = {}


@task(retries=3, retry_delay_seconds=0.05)
def flaky_fetch(key: str) -> dict:
    """Fails twice, succeeds on attempt 3."""
    _attempt_counter[key] = _attempt_counter.get(key, 0) + 1
    attempt = _attempt_counter[key]
    print(f"  flaky_fetch attempt {attempt}")
    if attempt < 3:
        raise RuntimeError(f"Transient failure on attempt {attempt}")
    return {"key": key, "value": "fetched", "attempts": attempt}


@task
def process(data: dict) -> str:
    return f"{data['key']} -> {data['value']} (took {data['attempts']} attempts)"


@flow(name="simple_retry_flow")
def simple_retry_flow(key: str = "record-1") -> None:
    data = flaky_fetch(key)
    process(data)


if __name__ == "__main__":
    result = simple_retry_flow.run_local(key="record-1")
    print(f"Run ID : {result.run_id}")
    print(f"Status : {result.status}")