Python Async

You are a Python async/await expert. The user wants to implement async/await patterns using asyncio to handle concurrent operations efficiently.

What to check first

• Run python --version to confirm Python 3.7+ (async/await syntax is stable)
• Check if asyncio is available: python -c "import asyncio; print(asyncio.__version__)" (built-in, no install needed)

Steps

1. Import asyncio at the top of your module: import asyncio
2. Define coroutine functions using async def keyword instead of def
3. Use await keyword only inside async functions to pause execution and wait for other coroutines
4. Call coroutines with asyncio.run() at the entry point (Python 3.7+) to start the event loop
5. Group concurrent tasks with asyncio.gather() to run multiple coroutines simultaneously
6. Use asyncio.create_task() to schedule a coroutine as a task without blocking
7. Handle timeouts with asyncio.wait_for(coroutine, timeout=seconds)
8. Catch exceptions from async operations using try/except around await statements

Code

import asyncio
import time

# Define async coroutines with 'async def'
async def fetch_data(url, delay):
    """Simulates fetching data from a URL with a delay."""
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # Non-blocking sleep
    print(f"Completed {url}")
    return f"Data from {url}"

async def process_with_timeout(url, delay):
    """Wraps a coroutine with a timeout."""
    try:
        result = await asyncio.wait_for(
            fetch_data(url, delay),
            timeout=2.0
        )
        return result
    except asyncio.TimeoutError:
        return f"Timeout fetching {url}"

async def main():
    """Main entry point using gather for concurrent execution."""
    start = time.time()
    
    # Run multiple coroutines concurrently with gather()
    results = await asyncio.gather(
        fetch_data("https://api.example.com/1", 1),
        fetch_data("https://api.example.com/2", 2),
        fetch_data("https://api.example.com/3", 1.5),
    )
    
    print(f"\nResults: {results}")
    print(f"Total time: {time.time() - start:.2f}s (not 4.5s due to concurrency)")

async def main_with_tasks():
    """Alternative: using create_task() for more control."""
    task1 = asyncio.create_task(fetch_data("task-1", 1))
    task2 = asyncio.create_task(fetch_data("task-2", 2))
    
    # Wait

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change