Async Patterns & Best Practices

import aiohttp
import asyncio

# Async context manager ensures proper cleanup
async def fetch_data(url):
    # aiohttp session is an async context manager
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            return data
    # Session automatically closed here, even if error occurs

async def main():
    result = await fetch_data(
        'https://jsonplaceholder.typicode.com/todos/1'
    )
    print(result)

asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager

# Method 1: Using __aenter__ and __aexit__
class AsyncDatabaseConnection:
    def __init__(self, host):
        self.host = host
        self.connection = None

    async def __aenter__(self):
        """Called when entering 'async with' block"""
        print(f"Connecting to {self.host}...")
        await asyncio.sleep(1)  # Simulate connection
        self.connection = f"Connected to {self.host}"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Called when exiting 'async with' block"""
        print(f"Closing connection to {self.host}...")
        await asyncio.sleep(0.5)  # Simulate cleanup
        self.connection = None
        return False  # Don't suppress exceptions

# Usage
async def main():
    async with AsyncDatabaseConnection("localhost") as db:
        print(f"Using: {db.connection}")
        # Do work with connection
    # Connection automatically closed

# Method 2: Using @asynccontextmanager decorator
@asynccontextmanager
async def database_connection(host):
    """Simpler way to create async context manager"""
    # Setup
    print(f"Connecting to {host}...")
    connection = await connect_to_database(host)

    try:
        yield connection  # Provide resource
    finally:
        # Cleanup (always runs)
        print(f"Closing connection to {host}...")
        await connection.close()

# Usage
async def query_database():
    async with database_connection("localhost") as conn:
        result = await conn.query("SELECT * FROM users")
        return result

asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager

class AsyncConnectionPool:
    def __init__(self, max_connections=10):
        self.max_connections = max_connections
        self.available = []
        self.in_use = set()
        self.semaphore = asyncio.Semaphore(max_connections)
        self._next_id = 0

    async def create_connection(self):
        """Create a new connection"""
        await asyncio.sleep(0.1)  # Simulate connection time
        # Use a counter, not id(object()): the temporary object is freed
        # immediately and CPython reuses the address, so every connection
        # would end up with the same name and the set below would collapse
        self._next_id += 1
        return f"Connection-{self._next_id}"

    @asynccontextmanager
    async def acquire(self):
        """Acquire a connection from the pool"""
        async with self.semaphore:
            # Get existing or create new
            if self.available:
                conn = self.available.pop()
            else:
                conn = await self.create_connection()

            self.in_use.add(conn)

            try:
                yield conn
            finally:
                # Return to pool
                self.in_use.remove(conn)
                self.available.append(conn)

# Usage
async def worker(pool, worker_id):
    async with pool.acquire() as conn:
        print(f"Worker {worker_id} using {conn}")
        await asyncio.sleep(1)  # Simulate work
        print(f"Worker {worker_id} done")

async def main():
    pool = AsyncConnectionPool(max_connections=3)

    # 10 workers sharing 3 connections
    await asyncio.gather(*[
        worker(pool, i) for i in range(10)
    ])

asyncio.run(main())

import asyncio

# Async generator function
async def fetch_pages(num_pages):
    """Simulate fetching pages one at a time"""
    for page in range(1, num_pages + 1):
        print(f"Fetching page {page}...")
        await asyncio.sleep(1)  # Simulate API call
        yield {"page": page, "data": f"Content of page {page}"}

# Consuming with async for
async def main():
    async for page in fetch_pages(5):
        print(f"Processing: {page}")
        # Process each page as it arrives
        # Memory efficient - one page at a time

asyncio.run(main())

import asyncio
import aiohttp

# Method 1: Using __aiter__ and __anext__
class AsyncAPIIterator:
    def __init__(self, base_url, max_pages):
        self.base_url = base_url
        self.max_pages = max_pages
        self.current_page = 1

    def __aiter__(self):
        """Return the iterator object (self)"""
        return self

    async def __anext__(self):
        """Return the next item or raise StopAsyncIteration"""
        if self.current_page > self.max_pages:
            raise StopAsyncIteration

        # Fetch page asynchronously
        url = f"{self.base_url}?_page={self.current_page}&_limit=2"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.json()

        self.current_page += 1
        return data

# Method 2: Async Generator (simpler)
async def paginated_api(base_url, max_pages):
    """Async generator - cleaner syntax"""
    for page in range(1, max_pages + 1):
        async with aiohttp.ClientSession() as session:
            url = f"{base_url}?_page={page}&_limit=2"
            async with session.get(url) as response:
                data = await response.json()
                yield data

# Usage: both produce the same pages, so print the ids to keep it short
API = "https://jsonplaceholder.typicode.com/posts"

async def main():
    print("Class-based iterator:")
    async for page_data in AsyncAPIIterator(API, 3):
        print(f"  page ids: {[row['id'] for row in page_data]}")

    print("Async generator:")
    async for page_data in paginated_api(API, 3):
        print(f"  page ids: {[row['id'] for row in page_data]}")

asyncio.run(main())

# A throwaway PostgreSQL. The user, password and database name match the
# DSN in the example below, so it runs without editing anything.
docker run -d --name asyncpg-demo \
    -e POSTGRES_USER=user \
    -e POSTGRES_PASSWORD=pass \
    -e POSTGRES_DB=db \
    -p 5432:5432 \
    postgres:16-alpine

# The first start takes a few seconds, so wait for it rather than guessing.
# Run pg_isready straight after docker run and it answers "no response".
until docker exec asyncpg-demo pg_isready -U user -d db > /dev/null 2>&1; do
    sleep 1
done

docker exec asyncpg-demo pg_isready -U user -d db

pip install asyncpg

# When you are finished, throw it away
docker rm -f asyncpg-demo

import asyncio
import asyncpg

DSN = 'postgresql://user:pass@localhost/db'

async def stream_large_table(pool, batch_size=1000):
    """Stream millions of rows without loading all into memory"""
    async with pool.acquire() as conn:
        # A cursor keeps the rows on the server; only batch_size at a
        # time crosses into Python. It has to run inside a transaction.
        async with conn.transaction():
            cursor = await conn.cursor(
                "SELECT * FROM large_table ORDER BY id"
            )

            while True:
                # Fetch batch
                rows = await cursor.fetch(batch_size)
                if not rows:
                    break

                # Yield batch for processing
                for row in rows:
                    yield row

async def process_record(record):
    """Stand-in for whatever real work happens per row"""
    return record['id']

# Usage
async def process_records():
    pool = await asyncpg.create_pool(DSN)

    processed = 0
    async for record in stream_large_table(pool):
        # Process one record at a time
        await process_record(record)
        processed += 1

        if processed % 10000 == 0:
            print(f"Processed {processed} records...")

    await pool.close()
    print(f"Total processed: {processed}")

asyncio.run(process_records())

import asyncio
import aiohttp

async def fetch_with_error_handling(url):
    """Proper error handling in async function"""
    try:
        timeout = aiohttp.ClientTimeout(total=10)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get(url) as response:
                response.raise_for_status()  # Raise for 4xx/5xx
                return await response.json()

    except asyncio.TimeoutError:
        print(f"Timeout fetching {url}")
        return None

    except aiohttp.ClientError as e:
        print(f"HTTP error for {url}: {e}")
        return None

    except Exception as e:
        print(f"Unexpected error for {url}: {e}")
        return None

    finally:
        # Cleanup code (always runs)
        print(f"Finished attempt for {url}")

# Usage: this id does not exist, so the API answers 404 and
# raise_for_status() turns that into a ClientError
async def main():
    result = await fetch_with_error_handling(
        "https://jsonplaceholder.typicode.com/posts/99999"
    )
    if result:
        print(f"Success: {result}")
    else:
        print("Failed to fetch data")

asyncio.run(main())

import asyncio

async def risky_task(task_id):
    """Task that might fail"""
    await asyncio.sleep(1)
    if task_id == 2:
        raise ValueError(f"Task {task_id} failed!")
    return f"Task {task_id} succeeded"

# Method 1: gather() with return_exceptions
async def main_gather():
    results = await asyncio.gather(
        risky_task(1),
        risky_task(2),  # This will fail
        risky_task(3),
        return_exceptions=True  # Don't crash, return exceptions
    )

    for i, result in enumerate(results, 1):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i}: {result}")

# Method 2: Wrapping tasks individually
async def safe_task(task_id):
    """Wrapper with error handling"""
    try:
        return await risky_task(task_id)
    except Exception as e:
        print(f"Caught error in task {task_id}: {e}")
        return None  # Return default value

async def main_wrapped():
    results = await asyncio.gather(
        safe_task(1),
        safe_task(2),  # Error caught internally
        safe_task(3)
    )
    print(f"All results: {results}")

# Method 3: Using create_task with error handling
async def main_tasks():
    tasks = [
        asyncio.create_task(safe_task(i))
        for i in range(1, 4)
    ]

    results = await asyncio.gather(*tasks)
    print(f"Results: {results}")

asyncio.run(main_gather())

import asyncio
import random

async def retry_with_backoff(
    coro_func,
    max_retries=3,
    base_delay=1,
    max_delay=60,
    exponential_base=2
):
    """Retry with exponential backoff"""
    for attempt in range(max_retries):
        try:
            return await coro_func()

        except Exception as e:
            if attempt == max_retries - 1:
                # Last attempt, re-raise
                raise

            # Calculate delay with exponential backoff
            delay = min(
                base_delay * (exponential_base ** attempt),
                max_delay
            )

            # Add jitter to prevent thundering herd
            jitter = random.uniform(0, delay * 0.1)
            total_delay = delay + jitter

            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {total_delay:.2f}s...")

            await asyncio.sleep(total_delay)

# Usage
# Seeded so this example prints the same thing every run. Drop the seed
# and you get a different number of retries each time, which is exactly
# what the backoff logic is there to handle.
random.seed(10)

async def unreliable_api_call():
    """Simulates unreliable API"""
    if random.random() < 0.7:
        raise Exception("API temporarily unavailable")
    return {"data": "success"}

async def main():
    try:
        result = await retry_with_backoff(
            unreliable_api_call,
            max_retries=5,
            base_delay=1
        )
        print(f"Success: {result}")
    except Exception as e:
        print(f"Failed after all retries: {e}")

asyncio.run(main())

import asyncio
import pytest

# Using pytest-asyncio. Keep the plain async test off the network: the
# mocked test below is how you cover the code that does make requests.
@pytest.mark.asyncio
async def test_async_function():
    """Test async function with pytest"""
    result = await task(3)
    assert result == 30

# Testing with mocks
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_with_mock():
    """Test with mocked async calls"""
    mock_response = AsyncMock()
    mock_response.json.return_value = {"data": "test"}

    # session.get(url) is used as an async context manager, so the code
    # under test receives __aenter__'s return value, not get()'s. Patching
    # get(return_value=mock_response) leaves .json() on the wrong object
    # and the assertion then compares against a MagicMock.
    with patch('aiohttp.ClientSession.get') as mock_get:
        mock_get.return_value.__aenter__.return_value = mock_response

        result = await fetch_data("https://api.example.com")
        assert result["data"] == "test"

# Testing timeouts
@pytest.mark.asyncio
async def test_timeout():
    """Test that operation times out"""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(), timeout=1.0)

# Testing concurrent operations
@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test multiple concurrent operations"""
    start = asyncio.get_running_loop().time()

    results = await asyncio.gather(
        task(1), task(2), task(3)
    )

    elapsed = asyncio.get_running_loop().time() - start

    assert len(results) == 3
    assert elapsed < 2.0  # Should run concurrently

# Fixture for async setup/teardown
@pytest.fixture
async def db_connection():
    """Async fixture"""
    conn = await create_connection()
    yield conn
    await conn.close()

@pytest.mark.asyncio
async def test_with_fixture(db_connection):
    """Use async fixture"""
    result = await db_connection.query("SELECT 1")
    assert result is not None