Retry Condition

"""
05_retry_condition.py: Selective retry based on exception type.

Demonstrates:
- retry_condition_fn=lambda e: isinstance(e, ValueError)
- Only ValueError triggers a retry; other exceptions propagate immediately
"""

from dagy import flow, task

_counts: dict[str, int] = {}


def only_value_errors(exc: Exception) -> bool:
    """Return True only for transient ValueError. Skip retry for other errors."""
    return isinstance(exc, ValueError)


@task(retries=3, retry_delay_seconds=0.02, retry_condition_fn=only_value_errors)
def validate_and_fetch(item_id: str) -> dict:
    """Raises ValueError on first attempt (will retry), succeeds on second."""
    _counts[item_id] = _counts.get(item_id, 0) + 1
    n = _counts[item_id]
    print(f"  validate_and_fetch({item_id!r}) attempt #{n}")
    if n == 1:
        raise ValueError(f"Validation failed for {item_id} - will retry")
    return {"item_id": item_id, "data": "valid", "attempts": n}


@task
def persist(record: dict) -> str:
    return f"Saved {record['item_id']} after {record['attempts']} attempt(s)"


@flow(name="retry_condition_flow")
def retry_condition_flow(item_id: str = "item-99") -> None:
    record = validate_and_fetch(item_id)
    persist(record)


if __name__ == "__main__":
    result = retry_condition_flow.run_local(item_id="item-99")
    print(f"Run ID : {result.run_id}")
    print(f"Status : {result.status}")