6.4Python Async Patterns

asyncio Basics

import asyncio
import httpx

async def test_concurrent_requests():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"https://api.example.com/items/{i}") for i in range(100)]
        responses = await asyncio.gather(*tasks)
        assert all(r.status_code == 200 for r in responses)

pytest-asyncio

import pytest
import httpx

@pytest.mark.asyncio
async def test_api_health(base_url):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{base_url}/health")
        assert response.status_code == 200

@pytest.mark.asyncio
async def test_parallel_endpoint_availability(base_url):
    endpoints = ["/users", "/products", "/orders", "/health"]
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"{base_url}{ep}") for ep in endpoints]
        responses = await asyncio.gather(*tasks)

    for endpoint, response in zip(endpoints, responses):
        assert response.status_code == 200, f"{endpoint} returned {response.status_code}"

asyncio.wait_for with Timeout

async def test_slow_endpoint_timeout():
    async with httpx.AsyncClient() as client:
        try:
            response = await asyncio.wait_for(
                client.get("https://api.example.com/slow-endpoint"),
                timeout=5.0
            )
            assert response.status_code == 200
        except asyncio.TimeoutError:
            pytest.fail("Endpoint did not respond within 5 seconds")