Sleep

from pyworkflow import workflow, sleep

@workflow()
async def reminder_sequence(user_id: str):
    await send_reminder(user_id, "First reminder")

    # Workflow suspends here - zero resources used
    await sleep("1d")

    # Resumes automatically after 1 day
    await send_reminder(user_id, "Second reminder")

    await sleep("7d")

    await send_reminder(user_id, "Final reminder")

Workflow Execution
        │
        ▼
┌───────────────┐
│ Execute Steps │
└───────┬───────┘
        │
        ▼
┌───────────────┐
│ sleep("1d")   │
└───────┬───────┘
        │
        ├─── 1. Record sleep_started event
        │
        ├─── 2. Schedule Celery Beat task for wake time
        │
        ├─── 3. Raise SuspensionSignal
        │
        └─── 4. Worker is freed
                  │
                  │  ... 1 day passes ...
                  │
                  ▼
        ┌─────────────────┐
        │ Celery Beat     │
        │ triggers resume │
        └────────┬────────┘
                 │
                 ▼
        ┌───────────────┐
        │ Replay events │
        │ Resume work   │
        └───────────────┘

await sleep("30s")   # 30 seconds
await sleep("5m")    # 5 minutes
await sleep("2h")    # 2 hours
await sleep("1d")    # 1 day
await sleep("1w")    # 1 week

# Combined
await sleep("1h30m") # 1 hour 30 minutes
await sleep("2d12h") # 2 days 12 hours

from datetime import timedelta

await sleep(timedelta(hours=2, minutes=30))
await sleep(timedelta(days=7))

from datetime import datetime

# Sleep until a specific datetime
await sleep(datetime(2025, 12, 25, 9, 0, 0))

# Sleep until next Monday at 9 AM
next_monday = get_next_monday()
await sleep(next_monday.replace(hour=9, minute=0))

await sleep(300)  # 300 seconds (5 minutes)

# BAD: This holds a worker for 24 hours!
import asyncio

async def traditional_sleep():
    await do_something()
    await asyncio.sleep(86400)  # Blocks worker for 24h
    await do_something_else()

# GOOD: Worker is freed during sleep
from pyworkflow import sleep

@workflow()
async def efficient_sleep():
    await do_something()
    await sleep("24h")  # Worker freed, resumes later
    await do_something_else()

@workflow()
async def onboarding_drip(user_id: str):
    await send_email(user_id, "Welcome!")

    await sleep("1d")
    await send_email(user_id, "Getting started tips")

    await sleep("3d")
    await send_email(user_id, "Advanced features")

    await sleep("7d")
    await send_email(user_id, "How's it going?")

@workflow()
async def process_refund(order_id: str):
    await validate_refund_request(order_id)

    # Wait for potential fraud review
    await sleep("24h")

    # If not flagged, process refund
    await execute_refund(order_id)
    await notify_customer(order_id)

@workflow()
async def batch_api_calls(items: list):
    for i, item in enumerate(items):
        await call_api(item)

        # Rate limit: 10 calls per minute
        if (i + 1) % 10 == 0:
            await sleep("1m")

@workflow()
async def resilient_operation():
    for attempt in range(5):
        try:
            result = await risky_step()
            return result
        except TemporaryError:
            if attempt < 4:
                # Exponential backoff: 1m, 2m, 4m, 8m
                delay = f"{2 ** attempt}m"
                await sleep(delay)

    raise FatalError("All retries exhausted")

@step(timeout="30s")  # Fails if step takes > 30s
async def quick_step():
    pass

@workflow()
async def my_workflow():
    await quick_step()
    await sleep("1h")  # Intentional 1-hour pause
    await quick_step()

# Start Celery Beat for scheduled task execution
celery -A pyworkflow.celery.app beat --loglevel=info

services:
  worker:
    command: celery -A pyworkflow.celery.app worker --loglevel=info

  beat:
    command: celery -A pyworkflow.celery.app beat --loglevel=info