Scheduling & retries

Run tasks on a cron schedule, fixed interval, or with automatic retry on failure.

Cron schedule

Use gw.Cron with standard 5-field UTC cron syntax. The task runs according to the schedule as long as it is deployed.

python
1@app.task(schedule=gw.Cron("0 */6 * * *"))  # every 6 hours2async def sync_data():3    pull_from_api()4    write_to_volume()

Fixed interval

gw.Period runs the task every N seconds. Simpler than cron for fixed-rate jobs.

python
1@app.task(schedule=gw.Period(seconds=300))  # every 5 minutes2async def heartbeat():3    ping_monitoring_service()

Retry policy

Automatically retry a task on failure with exponential backoff. max_retries is the total number of retry attempts (not including the first try).

python
1@app.task(2    retries=gw.Retries(3        max_retries=3,4        initial_delay_s=1.0,5        backoff_coefficient=2.0,  # 1s → 2s → 4s6    )7)8async def flaky_api_call(url: str) -> dict:9    import httpx10    r = httpx.get(url, timeout=10)11    r.raise_for_status()12    return r.json()

Attempt number

Read the current attempt inside the task body to skip expensive setup on retries.

python
1@app.task(retries=3)2async def process(item_id: str):3    attempt = gw.current_attempt_number()4    if attempt > 1:5        print(f"retry #{attempt} for {item_id}")6    do_work(item_id)

Integer retries shorthand

Pass an int to retries= for a simple fixed-count retry with default backoff.

python
1@app.task(retries=5)2async def resilient_task(x: int) -> int:3    return expensive_compute(x)