Asyncio for I/O-Heavy Pipelines: aiohttp, asyncpg, Rate Limiting, Backpressure

asyncio pipelines is the concurrency model every Python data engineer eventually reaches for when a pipeline has to fetch 10 K URLs from a partner API, poll 500 GraphQL endpoints per minute, fan out per-partition Postgres queries during a big backfill, or serve 100 K webhook deliveries per hour without spawning a thread per request and blowing memory on stack frames. Every DE eventually writes an async scraper or an async DB fanout; knowing when to reach for asyncio.gather versus as_completed versus 3.11+ TaskGroup, how to bound concurrency with Semaphore so you don't hammer a partner API into rate-limit errors, how to build backpressure with a bounded Queue so upstream fetchers pause when downstream sinks are slow, and how to compose it all under asyncio.timeout() for deadlines with clean cancellation is what separates a mid-level Python engineer from a senior one.

The tour walks the five pillars — (1) aiohttp.ClientSession with TCPConnector(limit=100, limit_per_host=10) for HTTP scraping plus timeout composition and DNS caching, (2) asyncpg for Postgres async with connection pools, prepared statements, and copy_records_to_table for bulk load (3-10× faster than psycopg async), (3) rate limiting via asyncio.Semaphore(N) for concurrency bounds and aiolimiter.AsyncLimiter(rate, period) for leaky-bucket rps caps plus bounded asyncio.Queue(maxsize=N) for backpressure, (4) production patterns covering gather(*tasks, return_exceptions=True) vs as_completed() vs 3.11+ TaskGroup for structured concurrency with automatic cancellation on error, and (5) the gotchas that bite every asyncio codebase — time.sleep inside coroutines, sync DB drivers on the event loop, task-cancellation semantics, and the one-event-loop-per-process rule. Every section ships a Solution-Tail interview answer — code, trace, output, why-this-works with __concept__ underlines.

Practice on SQL library →, SQL optimization drills →, and SQL join drills →.

---

On this page

• Why asyncio matters for pipelines in 2026
• aiohttp for HTTP scraping
• asyncpg for Postgres async
• Rate limiting + backpressure
• Patterns + gotchas + dialect matrix
• Cheat sheet — asyncio recipe list
• Frequently asked questions
• Practice on PipeCode

---

1. Why asyncio matters for pipelines in 2026

The asyncio pipelines mental model — one event loop, thousands of coroutines, bounded concurrency, structured cancellation

The one-sentence invariant: asyncio uses a single OS thread with a cooperative event loop to multiplex thousands of concurrent I/O operations — a coroutine awaits on a socket read, yields control back to the loop while the OS handles the I/O, and the loop resumes it when the socket is ready — giving you thread-count-independent concurrency, no per-task stack cost, and predictable ordering; the trade-off is that any time.sleep() or sync DB driver on the event loop stops every coroutine at once. Every high-throughput I/O-bound pipeline in modern Python is built on this model.

Where asyncio actually shows up in DE pipelines.

• Bulk HTTP scraping. 10 K URLs from a partner catalog; fetch them with 20 concurrent workers behind a Semaphore.
• Webhook fanout. Notify 100 downstream services per event; parallel POSTs with retries.
• API polling. Poll 500 GraphQL endpoints every minute; asyncio keeps memory footprint tiny.
• Fan-out per-partition DB queries. Backfill 128 partitions concurrently via asyncpg connection pool.
• CDC apply-loops. Consume Kafka async; write to Postgres async; both hands on the same event loop.
• Async ETL orchestration. Prefect / Dagster tasks that fan out sub-tasks via asyncio.
• FastAPI backends. Every route handler is a coroutine; DB / Redis / HTTP calls all await.
• Streaming pipelines. aiokafka consumer / producer for Kafka.

Where asyncio doesn't help.

• CPU-bound work. GIL still applies; use multiprocessing or a compiled extension.
• Low concurrency (< 10 tasks). Threading is simpler; the async ceremony isn't worth it.
• Single-shot I/O. A one-off script fetching one URL — plain requests / psycopg is fine.
• Libraries with no async driver. SQL Server via pyodbc — no async path; threading is the workaround.

What senior interviewers probe.

• When async vs threads vs processes. I/O-bound / high concurrency / need many tasks → async.
• gather vs as_completed vs TaskGroup. Wait-all vs stream vs structured concurrency.
• How to bound concurrency. Semaphore(N) — never let 10 K coroutines all try DB or HTTP.
• How to rate-limit. aiolimiter.AsyncLimiter(100, 60) — 100 requests per minute.
• Backpressure. Queue(maxsize=1000) — producer blocks when consumer is slow.
• Cancellation. task.cancel() + except CancelledError for cleanup.
• Timeouts. async with asyncio.timeout(10): — 3.11+ syntax.
• The forbidden calls. time.sleep(), blocking requests.get, sync psycopg on the loop.
• Event loop model. Single-threaded; concurrent I/O; not parallel CPU.

The three-line skeleton every DE writes.

import asyncio

async def main():
    results = await asyncio.gather(*[fetch(u) for u in urls])
    return results

results = asyncio.run(main())

The four asyncio anti-patterns.

• Blocking call on the loop. time.sleep(1) → freezes every coroutine.
• Sync driver in async code. psycopg2.connect(...).execute() on the event loop.
• Unbounded fanout. asyncio.gather(*[fetch(u) for u in 100_000_urls]) — creates 100 K coroutines and hammers the server.
• Sharing session/pool across processes. ClientSession bound to loop; forking multiplies bugs.

Worked example — a naive scraper vs a bounded, timed, session-reused scraper

Detailed explanation. A junior writes a scraper that opens a new HTTP session per URL, has no timeout, no concurrency bound, and no error handling. The pipeline hits ~50 URLs/sec, hammers the partner API, and eventually dies of connection-limit errors. The senior version reuses ClientSession, bounds concurrency to 20, sets timeouts, retries transiently, and hits 5 K URLs/sec smoothly.

Question. Rewrite the naive scraper into a production-ready version.

Input. 10,000 URLs to fetch; partner tolerates 100 rps and 10 concurrent connections per host.

Code.

# NAIVE — will crash under load
import aiohttp, asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as s:   # new session per URL
        async with s.get(url) as r:
            return await r.text()

async def main(urls):
    return await asyncio.gather(*[fetch(u) for u in urls])   # unbounded

# PRODUCTION — session reused, bounded, timed, retryable
async def fetch_one(session, url, sem):
    async with sem:
        for attempt in range(3):
            try:
                async with session.get(url) as r:
                    r.raise_for_status()
                    return await r.text()
            except (aiohttp.ClientError, asyncio.TimeoutError):
                if attempt == 2:
                    raise
                await asyncio.sleep(0.5 * 2 ** attempt)

async def main(urls):
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10, ttl_dns_cache=300)
    timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
    sem = asyncio.Semaphore(20)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        return await asyncio.gather(
            *[fetch_one(session, u, sem) for u in urls],
            return_exceptions=True,
        )

Step-by-step explanation.

1. Naive: new ClientSession per URL — TCP handshake + DNS lookup every time; ~50 URLs/sec ceiling.
2. Production: one ClientSession reused across all 10 K URLs — connection pool amortises TCP+DNS.
3. TCPConnector(limit=100, limit_per_host=10) — total 100 open connections; max 10 per host; respects partner limits.
4. ttl_dns_cache=300 — DNS resolved once per host per 5 min.
5. ClientTimeout(total=30, connect=5, sock_read=10) — layered deadlines.
6. Semaphore(20) — no more than 20 coroutines actually inside session.get at a time.
7. Retry loop with exponential backoff — 0.5 s → 1 s → 2 s.
8. return_exceptions=True — one URL's failure doesn't cancel the other 9,999.

Output.

Metric	Naive	Production
URLs/sec	~50	~5,000
TCP handshakes	10,000	~50
Errors on server	frequent (connection limit)	rare
Failure mode	worker crashes	per-URL exception captured

Rule of thumb. Every production async scraper needs: one ClientSession shared, TCPConnector limits, layered timeouts, Semaphore for concurrency bound, return_exceptions=True on gather, retry with backoff.

Worked example — the sync-call-on-loop bug

Detailed explanation. A coroutine calls a synchronous library (requests.get, time.sleep, psycopg2.execute) which blocks the event loop for the duration of the call. Every other coroutine stops. The whole app appears to hang.

Question. Show the bug and the two fixes.

Code.

import time, requests, asyncio

# BUGGY
async def worker(url):
    time.sleep(1)           # BLOCKS THE LOOP
    r = requests.get(url)   # ALSO BLOCKS
    return r.text

# FIX 1 — use async equivalents
async def worker(url):
    await asyncio.sleep(1)
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            return await r.text()

# FIX 2 — offload to threadpool if no async equivalent
async def worker(url):
    await asyncio.sleep(1)
    return await asyncio.to_thread(requests.get, url).text

Rule of thumb. No sync I/O and no time.sleep in coroutines. Either find an async driver or use asyncio.to_thread (3.9+) to offload.

Worked example — 3.11+ TaskGroup for structured concurrency

Detailed explanation. Prior to 3.11, asyncio.gather didn't have great cancellation semantics — one task failing didn't clean up siblings by default. TaskGroup fixes this.

Code.

async def fetch_all(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    # On exit, all tasks awaited; if any raised, all others cancelled;
    # exception raised as ExceptionGroup.
    return [t.result() for t in tasks]

Rule of thumb. In 3.11+, prefer TaskGroup over gather for fail-fast semantics with automatic sibling cancellation.

Common beginner mistakes

• Blocking calls on the event loop (time.sleep, requests.get, sync psycopg).
• New ClientSession per request instead of reusing.
• Unbounded fanout: gather(*[fetch(u) for u in 10_000_000_urls]).
• Not setting timeouts — one slow URL hangs the pipeline.
• Missing return_exceptions=True — one failure cancels all.
• Sharing ClientSession or Pool across processes.
• Running asyncio.run(main()) inside an already-running loop (Jupyter, FastAPI).

asyncio pipelines interview question on refactoring a threaded scraper

A senior interviewer often asks: "You inherit a threaded scraper using ThreadPoolExecutor(max_workers=50) that consumes 4 GB RAM and does 200 URLs/sec. Rewrite in asyncio with the same or better throughput, less memory."

Solution Using asyncio with ClientSession + Semaphore + retry + backpressure

import asyncio, aiohttp
from asyncio import Semaphore, Queue

async def worker(session, sem, in_q, out_q):
    while True:
        url = await in_q.get()
        if url is None:
            in_q.task_done()
            break
        async with sem:
            for attempt in range(3):
                try:
                    async with session.get(url) as r:
                        r.raise_for_status()
                        text = await r.text()
                    await out_q.put((url, text, None))
                    break
                except Exception as e:
                    if attempt == 2:
                        await out_q.put((url, None, str(e)))
                    else:
                        await asyncio.sleep(0.5 * 2 ** attempt)
        in_q.task_done()

async def main(urls, num_workers=50):
    connector = aiohttp.TCPConnector(limit=200, limit_per_host=20, ttl_dns_cache=300)
    timeout = aiohttp.ClientTimeout(total=30, connect=5)
    sem = Semaphore(50)
    in_q: Queue = Queue(maxsize=1000)
    out_q: Queue = Queue()

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        # Spawn workers
        workers = [
            asyncio.create_task(worker(session, sem, in_q, out_q))
            for _ in range(num_workers)
        ]
        # Feed URLs with backpressure
        for url in urls:
            await in_q.put(url)
        # Signal end
        for _ in workers:
            await in_q.put(None)
        # Wait for completion
        await asyncio.gather(*workers)

    # Drain results
    results = []
    while not out_q.empty():
        results.append(await out_q.get())
    return results

Step-by-step trace.

Component	Behavior
in_q with maxsize=1000	Backpressure — producer blocks if queue full
50 worker coroutines	Consume from in_q, fetch, push to out_q
Semaphore(50)	No more than 50 concurrent HTTP calls
Retry with backoff	0.5s → 1s → 2s
Sentinel None	Signals worker to exit
gather(*workers)	Waits for all workers to drain and exit

Output:

Metric	Threaded	Asyncio
Throughput	200 URLs/sec	500-1000 URLs/sec
Memory	4 GB (50 thread stacks + I/O buffers)	200 MB (single loop)
Setup complexity	ThreadPoolExecutor + Session	ClientSession + Semaphore + Queue
Error isolation	per-thread	per-coroutine

Why this works — concept by concept:

• One ClientSession shared — connection pool amortises TCP handshakes across all URLs.
• Semaphore(50) — bounds concurrent HTTP calls to what the server tolerates.
• Bounded Queue(maxsize=1000) — backpressure: if consumers slow, producer blocks at put, preventing memory explosion.
• Sentinel-based worker termination — each worker sees None and exits cleanly; no cancellation errors.
• Retry with exponential backoff — transient errors self-heal; permanent errors surface after 3 attempts.
• Cost — memory dominated by connection pool + in-flight buffers, not per-task overhead; typical 200-400 MB for 50 concurrent HTTP calls.

SQL
Topic — SQL
SQL practice library

Practice →

SQL Topic — optimization SQL optimization drills

Practice →

---

2. aiohttp for HTTP scraping

aiohttp.ClientSession — the async HTTP client every Python DE uses

The mental model in one line: aiohttp.ClientSession is the async equivalent of the requests.Session — reuse it across all requests to amortise TCP handshake + TLS + DNS costs, configure the connector to bound total concurrency and per-host limits (so you don't hammer a partner API), layer timeouts to prevent one hanging URL from blocking the pipeline, and always close the session cleanly via async with context manager to avoid connection leaks.

Slot 1 — the canonical ClientSession setup.

connector = aiohttp.TCPConnector(
    limit=100,               # total open connections
    limit_per_host=10,       # per-host cap
    ttl_dns_cache=300,       # 5 min DNS cache
    enable_cleanup_closed=True,
    force_close=False,       # keep-alive
)

timeout = aiohttp.ClientTimeout(
    total=30,                # overall deadline
    connect=5,               # TCP handshake
    sock_connect=5,          # socket connect
    sock_read=10,            # per-read
)

async with aiohttp.ClientSession(
    connector=connector,
    timeout=timeout,
    headers={"User-Agent": "myco-scraper/1.0"},
) as session:
    ...

Slot 2 — TCPConnector parameters explained.

• limit=N — max total simultaneous connections. Default 100.
• limit_per_host=N — max per-host simultaneous. Default 0 (unlimited).
• ttl_dns_cache=N — DNS entries cached for N seconds.
• use_dns_cache=True — cache DNS results.
• force_close=False — keep-alive; reuse connection for subsequent requests to same host.
• enable_cleanup_closed=True — close TLS connections gracefully.
• ssl=False — disable SSL verification (development only).

Slot 3 — ClientTimeout layers.

• total — overall timeout for the whole request (connect + send + read).
• connect — connecting to the pool; waits for a free connector slot.
• sock_connect — TCP socket connect only.
• sock_read — timeout between reading each chunk.
• Set total always; connect + sock_read layers as needed.

Slot 4 — request patterns.

# GET
async with session.get(url) as r:
    text = await r.text()
    j = await r.json()
    b = await r.read()

# POST with JSON body
async with session.post(url, json={"k": "v"}) as r:
    ...

# POST with form data
async with session.post(url, data={"k": "v"}) as r:
    ...

# Custom headers per request (in addition to session defaults)
async with session.get(url, headers={"Authorization": "Bearer X"}) as r:
    ...

# Query parameters
async with session.get(url, params={"page": 1, "limit": 100}) as r:
    ...

Slot 5 — error handling patterns.

• ClientResponse.raise_for_status() — raises ClientResponseError on 4xx/5xx.
• Retry only on transient. 5xx yes; 4xx (client error) no.
• ClientConnectorError — DNS / connection refused.
• asyncio.TimeoutError — deadline exceeded.
• ClientPayloadError — response truncated.

Slot 6 — aiohttp-retry for automatic retries.

from aiohttp_retry import RetryClient, ExponentialRetry

retry_options = ExponentialRetry(attempts=3, start_timeout=0.5)
async with RetryClient(retry_options=retry_options) as client:
    async with client.get(url) as r:
        ...

Handles transient errors + exponential backoff automatically.

Slot 7 — streaming large responses.

async with session.get(url) as r:
    async for chunk in r.content.iter_chunked(8192):
        process(chunk)

Avoids materializing the full body in memory.

Slot 8 — cookies + sessions.

# Automatic cookie jar
async with aiohttp.ClientSession() as s:
    async with s.post(login_url, data=creds) as r:
        pass   # cookies stored automatically
    async with s.get(protected_url) as r:
        pass   # cookies sent

# Manual cookie jar
jar = aiohttp.CookieJar()
async with aiohttp.ClientSession(cookie_jar=jar) as s:
    ...

Slot 9 — WebSocket via aiohttp.

async with session.ws_connect(url) as ws:
    await ws.send_str("hello")
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            print(msg.data)

Slot 10 — proxy support.

async with session.get(url, proxy="http://proxy.example.com:8080") as r:
    ...

Common beginner mistakes

• New ClientSession per request instead of shared session.
• Missing async with context — connection leak.
• No timeout — hangs forever on slow URL.
• limit_per_host=0 and hammering a single API.
• Not calling raise_for_status — silent 4xx/5xx.
• Retrying 4xx errors (bad request will always be bad).
• Not decoding response before session close.
• Sharing session across event loops / processes.

Worked example — a rate-limited paginated API scraper

Detailed explanation. Fetch all pages of a paginated JSON API that has next cursors and enforces 100 rps.

Question. Write the scraper.

Code.

import aiohttp, asyncio
from aiolimiter import AsyncLimiter

async def fetch_page(session, url, limiter):
    async with limiter:
        async with session.get(url) as r:
            r.raise_for_status()
            return await r.json()

async def fetch_all(base_url):
    connector = aiohttp.TCPConnector(limit=50, limit_per_host=10)
    timeout = aiohttp.ClientTimeout(total=30, connect=5)
    limiter = AsyncLimiter(100, 60)   # 100 rps
    all_items = []
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as s:
        url = base_url
        while url:
            page = await fetch_page(s, url, limiter)
            all_items.extend(page["data"])
            url = page.get("next_cursor")
    return all_items

Rule of thumb. Cursor pagination is inherently sequential; parallelism only helps with multi-source or multi-endpoint scraping.

Worked example — fanning out to 50 hosts respecting per-host limits

Detailed explanation. Fetch 10 K URLs across 50 hosts; enforce 10 concurrent per host, 100 total.

Code.

async def fetch(session, sem_by_host, url):
    host = url.split("/")[2]
    sem = sem_by_host.setdefault(host, asyncio.Semaphore(10))
    async with sem:
        async with session.get(url) as r:
            return await r.text()

Rule of thumb. Per-host semaphores let you enforce partner-specific rate limits precisely.

Worked example — streaming a large download without buffering

Code.

async with session.get(large_url) as r:
    with open("out.bin", "wb") as f:
        async for chunk in r.content.iter_chunked(65536):
            f.write(chunk)

Rule of thumb. Always stream large responses; never await r.read() for GB-scale files.

aiohttp interview question on session sharing

A senior interviewer asks: "Explain what happens when you create a new ClientSession per URL vs sharing one across the whole app. Give a benchmark and the mechanism."

Solution Using shared vs per-URL session benchmark

import time, aiohttp, asyncio

URLS = ["https://httpbin.org/get"] * 100

async def per_url():
    async def one(url):
        async with aiohttp.ClientSession() as s:
            async with s.get(url) as r:
                return await r.text()
    return await asyncio.gather(*[one(u) for u in URLS])

async def shared():
    async with aiohttp.ClientSession() as s:
        async def one(url):
            async with s.get(url) as r:
                return await r.text()
        return await asyncio.gather(*[one(u) for u in URLS])

t = time.perf_counter()
asyncio.run(per_url())
print(f"per-URL: {time.perf_counter()-t:.2f}s")

t = time.perf_counter()
asyncio.run(shared())
print(f"shared: {time.perf_counter()-t:.2f}s")

Output:

Approach	Wall clock (100 URLs)	Mechanism
Per-URL session	~15s	New TCP + TLS + DNS per URL
Shared session	~2s	Connection pool reuses

Why this works — concept by concept:

• Connection pool reuse — one TCP handshake amortised across many requests.
• DNS cache — one DNS lookup per host per TTL.
• TLS session resumption — cached TLS state.
• keep-alive — HTTP/1.1 or HTTP/2 multiplexing.
• Cost — per-URL cost drops from ~150ms overhead to ~5ms.

SQL
Topic — SQL
SQL practice library

Practice →

SQL Topic — joins SQL join drills

Practice →

---

3. asyncpg for Postgres async

asyncpg — the fastest Postgres async driver, 3-10× faster than psycopg async

The mental model in one line: asyncpg implements the Postgres wire protocol natively in Python + optimised C for async event-loop integration, giving 3-10× throughput over psycopg3 async on typical DE workloads; connection pooling via asyncpg.create_pool, prepared statements via conn.prepare, and copy_records_to_table for bulk load cover 95% of DE needs — the only major asymmetry vs psycopg is that asyncpg uses $1, $2 positional parameters (not %s).

Slot 1 — the pool pattern.

import asyncpg

async def get_pool():
    return await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5,
        max_size=20,
        max_inactive_connection_lifetime=300,   # 5 min
        command_timeout=30,
    )

# Usage
async def fetch_user(pool, user_id):
    async with pool.acquire() as conn:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

Slot 2 — fetch variants.

• await conn.fetch(sql, *args) — returns list of records.
• await conn.fetchrow(sql, *args) — returns first record or None.
• await conn.fetchval(sql, *args) — returns scalar value.
• await conn.execute(sql, *args) — runs statement, returns status string.
• await conn.executemany(sql, args_list) — batch execute.

Slot 3 — parameter placeholders.

# Positional $1, $2 — NOT %s
rows = await conn.fetch("SELECT * FROM users WHERE country = $1 AND age > $2", "US", 18)

Different from psycopg's %s; migration between drivers requires editing SQL.

Slot 4 — prepared statements.

async with pool.acquire() as conn:
    stmt = await conn.prepare("SELECT * FROM users WHERE id = $1")
    for user_id in user_ids:
        row = await stmt.fetchrow(user_id)
        process(row)

Parse once, execute many. Faster for tight loops.

Slot 5 — copy_records_to_table for bulk load.

records = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
async with pool.acquire() as conn:
    await conn.copy_records_to_table(
        "users",
        records=records,
        columns=["id", "name"],
    )

Uses Postgres COPY protocol; 10-100× faster than executemany.

Slot 6 — transactions.

async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("INSERT INTO orders ...")
        await conn.execute("UPDATE inventory ...")
        # commit on context exit; rollback on exception

Slot 7 — LISTEN/NOTIFY for pub/sub.

async with pool.acquire() as conn:
    await conn.add_listener("orders_channel", callback)
    while True:
        await asyncio.sleep(3600)   # keep connection alive

Slot 8 — array parameters.

rows = await conn.fetch(
    "SELECT * FROM users WHERE id = ANY($1::int[])",
    [1, 2, 3, 4, 5]
)

Slot 9 — JSON support.

# asyncpg supports jsonb natively
rows = await conn.fetch(
    "SELECT id, data->>'name' as name FROM events WHERE data->>'type' = $1",
    "click"
)

Slot 10 — asyncpg vs psycopg async performance.

Typical DE workload — 10 K SELECT + 1 K INSERT via prepared statement:

• asyncpg — ~5 sec.
• psycopg3 async — ~15-30 sec.
• psycopg2 (threaded) — ~20 sec.

Common beginner mistakes

• Using %s placeholders (works in psycopg, not asyncpg).
• Not using create_pool — creating connections per call.
• Missing async with pool.acquire() — connection leak.
• Using execute for bulk load — use copy_records_to_table.
• Sharing single connection across many coroutines (serialises them).
• Not setting command_timeout — one slow query hangs forever.
• Forgetting ::int[] cast on array parameters.

Worked example — bulk-load 1M records via COPY

Detailed explanation. Best throughput for bulk insert on Postgres is COPY. asyncpg's copy_records_to_table uses the wire protocol natively.

Question. Load 1M records fastest.

Code.

async def bulk_load(pool, records):
    async with pool.acquire() as conn:
        await conn.copy_records_to_table(
            "users",
            records=records,
            columns=["id", "name", "email"],
        )

# Compare to executemany
async def bulk_load_slow(pool, records):
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO users (id, name, email) VALUES ($1, $2, $3)",
            records
        )

Output:

Method	1M records
executemany	~90 sec
copy_records_to_table	~8 sec

Rule of thumb. For > 10K rows, use COPY. Executemany for smaller batches.

Worked example — fanning out reads with a pool

Code.

async def fetch_batch(pool, user_ids):
    async def one(uid):
        async with pool.acquire() as conn:
            return await conn.fetchrow("SELECT * FROM users WHERE id = $1", uid)
    return await asyncio.gather(*[one(uid) for uid in user_ids])

Each coroutine acquires a connection from the pool briefly.

Rule of thumb. Never hold a connection across long awaits — return to pool ASAP.

Worked example — prepared statement in a tight loop

Code.

async with pool.acquire() as conn:
    stmt = await conn.prepare("SELECT * FROM users WHERE id = $1")
    for uid in user_ids:   # 10K IDs
        row = await stmt.fetchrow(uid)
        process(row)

Rule of thumb. Prepared statements shave ~50% parse cost off repeated queries in a loop.

asyncpg interview question on pool sizing

A senior interviewer asks: "You have an app doing 1000 rps against Postgres via asyncpg. Each query ~10ms. What pool size?"

Solution Using Little's Law + margin

Concurrent-connections-needed = QPS × avg_query_time
                              = 1000 × 0.01
                              = 10 connections

With 50% headroom: max_size = 15-20 connections.

Pool config:
create_pool(min_size=5, max_size=20, command_timeout=30)

Rule of thumb. Little's Law: N = arrival_rate × service_time. Pool must accommodate the concurrent-connection count with headroom.

SQL
Topic — optimization
SQL optimization drills

Practice →

SQL Topic — SQL SQL practice library

Practice →

---

4. Rate limiting + backpressure

Semaphore for concurrency bound, aiolimiter for rps cap, Queue(maxsize=N) for backpressure

The mental model in one line: Semaphore(N) bounds how many coroutines are inside a critical section at once (concurrency); aiolimiter.AsyncLimiter(rate, period) implements a leaky-bucket that shapes throughput (requests per period); asyncio.Queue(maxsize=N) provides backpressure by blocking producers when the buffer is full — combine the three to keep a scraper polite (rate limit), efficient (concurrency), and memory-safe (backpressure).

Slot 1 — asyncio.Semaphore(N).

sem = asyncio.Semaphore(10)

async def worker(url):
    async with sem:   # blocks if > 10 already inside
        return await fetch(url)

# 1000 URLs, max 10 concurrent
await asyncio.gather(*[worker(u) for u in urls])

Slot 2 — aiolimiter.AsyncLimiter(max_rate, time_period).

from aiolimiter import AsyncLimiter

limiter = AsyncLimiter(100, 60)   # 100 requests per 60 seconds

async def fetch(url):
    async with limiter:
        return await http_get(url)

• Leaky-bucket rate limiter.
• Blocks when rate exceeded until tokens available.
• Precisely enforces API rate limits.

Slot 3 — bounded asyncio.Queue.

q = asyncio.Queue(maxsize=1000)

async def producer():
    for url in urls:
        await q.put(url)   # blocks if queue full
    for _ in range(N_WORKERS):
        await q.put(None)   # sentinels

async def consumer():
    while True:
        url = await q.get()
        if url is None:
            q.task_done()
            break
        result = await fetch(url)
        q.task_done()

Slot 4 — combined pattern.

async def worker(q, sem, results):
    while True:
        url = await q.get()
        if url is None:
            q.task_done()
            break
        async with sem:
            r = await fetch(url)
            results.append(r)
        q.task_done()

Semaphore bounds concurrency; Queue provides backpressure.

Slot 5 — token-bucket via aiometer.

Alternative library:

import aiometer
results = await aiometer.run_all(
    [functools.partial(fetch, u) for u in urls],
    max_at_once=10,
    max_per_second=100,
)

Combines Semaphore + rate limiter in one API.

Slot 6 — per-host semaphores.

sem_by_host = {}

def get_sem(url):
    host = url.split("/")[2]
    return sem_by_host.setdefault(host, asyncio.Semaphore(10))

async def worker(url):
    async with get_sem(url):
        return await fetch(url)

Slot 7 — burst allowance with aiolimiter.

limiter = AsyncLimiter(100, 60)
# Allows up to 100 immediate requests, then rate-limits

Under 100 requests in a burst: no delay. Above: paced at 100/60s.

Slot 8 — hierarchical rate limits.

global_limiter = AsyncLimiter(1000, 60)   # site-wide
per_endpoint = AsyncLimiter(100, 60)      # per endpoint

async def fetch(url):
    async with global_limiter, per_endpoint:
        ...

Slot 9 — dynamic rate adjustment.

Not built into aiolimiter; wrap manually:

class DynamicLimiter:
    def __init__(self, initial_rate, period):
        self.limiter = AsyncLimiter(initial_rate, period)

    def adjust(self, new_rate):
        self.limiter = AsyncLimiter(new_rate, self.limiter.time_period)

Slot 10 — the four combinations.

Constraint	Tool
Bound concurrency	Semaphore(N)
Cap rate (rps)	AsyncLimiter(rate, period)
Backpressure buffer	Queue(maxsize=N)
Rate + concurrency + backpressure	all three composed

Common beginner mistakes

• Semaphore without backpressure — memory blows up buffering URLs.
• Rate limit without concurrency bound — all 100 rps arrive as burst at 1 host.
• Unbounded Queue — producer runs faster than consumer; memory explodes.
• Forgetting sentinels — workers never exit; program hangs on shutdown.
• Rate-limiting inside Semaphore critical section (fine but paces the concurrency).

Worked example — scraper with concurrency + rate + backpressure

Code.

async def run_scraper(urls, num_workers=50, rps=100):
    sem = asyncio.Semaphore(20)
    limiter = AsyncLimiter(rps, 60)
    q: asyncio.Queue = asyncio.Queue(maxsize=1000)
    results = []

    async def worker():
        while True:
            url = await q.get()
            if url is None:
                q.task_done()
                return
            async with limiter, sem:
                try:
                    async with session.get(url) as r:
                        results.append(await r.text())
                except Exception as e:
                    print(f"error: {url}: {e}")
            q.task_done()

    connector = aiohttp.TCPConnector(limit=200)
    async with aiohttp.ClientSession(connector=connector) as session:
        workers = [asyncio.create_task(worker()) for _ in range(num_workers)]
        for u in urls:
            await q.put(u)
        for _ in workers:
            await q.put(None)
        await asyncio.gather(*workers)
    return results

Rule of thumb. Every production scraper composes Semaphore + rate limiter + bounded Queue.

Worked example — Postgres write pipeline with backpressure

Code.

async def kafka_to_postgres():
    q: asyncio.Queue = asyncio.Queue(maxsize=10_000)

    async def consumer():
        async for msg in kafka_consumer:
            await q.put(msg)   # backpressure

    async def writer():
        batch = []
        async with pg_pool.acquire() as conn:
            while True:
                msg = await q.get()
                batch.append(msg)
                if len(batch) >= 1000:
                    await conn.copy_records_to_table("events", records=batch)
                    batch.clear()
                q.task_done()

    await asyncio.gather(consumer(), writer())

Rule of thumb. Backpressure = writer sets pace; consumer blocks when writer slow.

asyncio interview question on rate + concurrency + backpressure interaction

A senior interviewer asks: "Design a scraper for 10 K URLs against a partner API that allows 100 rps and 10 concurrent connections per host. Explain how you compose the primitives."

Solution Using layered constraints

async def scrape(urls):
    connector = aiohttp.TCPConnector(limit_per_host=10)
    limiter = AsyncLimiter(100, 60)
    q = asyncio.Queue(maxsize=500)

    async def worker(session):
        while True:
            url = await q.get()
            if url is None:
                q.task_done()
                break
            async with limiter:   # rate limit
                async with session.get(url) as r:
                    result = await r.text()
            # limit_per_host handled by connector automatically
            q.task_done()

    async with aiohttp.ClientSession(connector=connector) as session:
        workers = [asyncio.create_task(worker(session)) for _ in range(20)]
        for u in urls:
            await q.put(u)
        for _ in workers:
            await q.put(None)
        await asyncio.gather(*workers)

Why this works — concept by concept:

• limit_per_host=10 on TCPConnector — the connector enforces 10 concurrent connections per host without needing a manual Semaphore per host.
• AsyncLimiter(100, 60) for rate — leaky bucket paces requests at exactly 100/60s.
• Queue(maxsize=500) for backpressure — producer blocks at put when workers are slow.
• 20 workers — enough to keep the pipeline full without wasting overhead.
• Sentinel pattern — clean worker shutdown.

SQL
Topic — optimization
SQL optimization drills

Practice →

SQL Topic — SQL SQL practice library

Practice →

---

5. Patterns + gotchas + dialect matrix

gather vs as_completed vs TaskGroup; cancellation; timeouts; one-loop rule

The mental model in one line: asyncio.gather(*tasks, return_exceptions=True) awaits all tasks in submission order (returns list); asyncio.as_completed(tasks) yields results as they finish (streams); 3.11+ TaskGroup provides structured concurrency with automatic sibling cancellation on error — combine with async with asyncio.timeout(sec) for deadline enforcement and clean except CancelledError for teardown; and remember: one event loop per process, no time.sleep on the loop, no sync DB drivers.

Slot 1 — asyncio.gather.

results = await asyncio.gather(
    fetch(url1),
    fetch(url2),
    fetch(url3),
    return_exceptions=True,
)
# results = [str_or_exception, str_or_exception, str_or_exception]
# Ordered same as input.

• Wait-all semantics.
• return_exceptions=True — one failure doesn't cancel others.

Slot 2 — asyncio.as_completed.

for coro in asyncio.as_completed([fetch(u) for u in urls]):
    result = await coro
    process(result)   # process as they finish

• Streams results in completion order.
• Better for processing large batches incrementally.

Slot 3 — 3.11+ TaskGroup.

async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(fetch(url1))
    task2 = tg.create_task(fetch(url2))
# On exit, both tasks awaited.
# If any raised, all others cancelled; exceptions raised as ExceptionGroup.

• Structured concurrency.
• Fail-fast semantics.
• Automatic cleanup.

Slot 4 — asyncio.timeout (3.11+).

async with asyncio.timeout(10):
    await long_operation()   # raises TimeoutError if > 10s

• Older syntax: asyncio.wait_for(coro, timeout=10).

Slot 5 — cancellation.

task = asyncio.create_task(long_op())
await asyncio.sleep(5)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("cancelled cleanly")

• task.cancel() schedules cancellation.
• Coroutine can catch CancelledError for cleanup.

Slot 6 — shielding.

await asyncio.shield(critical_operation())
# critical_operation runs to completion even if outer task cancelled

Slot 7 — synchronization primitives.

• Lock — mutex.
• Event — signal.
• Condition — condition variable.
• Semaphore — bounded permits.
• BoundedSemaphore — Semaphore that raises if .release() exceeds count.

Slot 8 — running blocking code.

# Python 3.9+
result = await asyncio.to_thread(blocking_func, arg1, arg2)

# Older syntax
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, blocking_func, arg)

• Offloads to thread pool.
• Use for third-party sync libraries (S3, GCS, etc. with sync SDKs).

Slot 9 — event loop rules.

• One loop per process. Never share across processes.
• asyncio.run() creates a fresh loop. Not composable with existing loop.
• In Jupyter / FastAPI — loop already running; use await directly.
• Fork after loop start — undefined behavior. Fork first, start loop in child.

Slot 10 — the 8-driver matrix.

Concern	Async library
HTTP client	aiohttp, httpx
HTTP server	aiohttp.web, FastAPI, Starlette
Postgres	asyncpg
MySQL	aiomysql
SQLite	aiosqlite
Redis	redis-py (async support)
Kafka	aiokafka, faust
RabbitMQ	aio-pika
S3	aiobotocore
MongoDB	motor

Common beginner mistakes

• Mixing sync and async in the same code path.
• asyncio.run() inside FastAPI or Jupyter (already-running loop).
• time.sleep(1) inside coroutine.
• requests.get() on the event loop.
• Not catching CancelledError — leaks resources.
• Not shielding critical cleanup.
• Sharing session/pool across forks.
• Setting create_task without keeping reference — task can be garbage collected.

Worked example — fail-fast fan-out with TaskGroup

async def fetch_all_or_fail(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    return [t.result() for t in tasks]

try:
    results = await fetch_all_or_fail(urls)
except* aiohttp.ClientError as eg:
    print(f"one or more fetches failed: {eg}")

Rule of thumb. TaskGroup for fail-fast; gather with return_exceptions for tolerate-partial.

Worked example — timeout composition

async def with_deadline():
    async with asyncio.timeout(30):
        async with asyncio.timeout(5):
            await fast_op()
        await slow_op()   # remaining 25s of outer budget

Rule of thumb. Nest timeouts for per-op deadlines within an overall budget.

Worked example — offloading sync SDK

# boto3 is sync; offload
async def upload_s3(key, body):
    return await asyncio.to_thread(boto3.client("s3").put_object, Key=key, Body=body)

Rule of thumb. When no async driver exists, asyncio.to_thread bridges without blocking the loop.

asyncio interview question on cancellation

A senior interviewer asks: "You're in a coroutine that acquires a lock, does I/O, releases the lock. It gets cancelled mid-I/O. Walk through what happens and how to ensure the lock is released."

Solution Using try/finally + shield

async def critical():
    lock = asyncio.Lock()
    await lock.acquire()
    try:
        await asyncio.shield(do_io())   # protect I/O from outer cancel
    finally:
        lock.release()

# Even cleaner: async with lock:
async def critical_v2():
    async with lock:
        await do_io()   # lock released on exception or cancel

Why this works — concept by concept:

• async with releases automatically — even on CancelledError.
• try/finally as fallback — for manual lock management.
• asyncio.shield — protects critical section from outer cancellation.
• CancelledError propagation — always re-raise unless you have specific cleanup.

SQL
Topic — SQL
SQL practice library

Practice →

SQL
Topic — optimization
SQL optimization drills

Practice →

---

Cheat sheet — asyncio recipe list

• async def + await — basics.
• asyncio.run(main()) — entry point.
• asyncio.gather(*tasks, return_exceptions=True) — parallel wait-all.
• asyncio.as_completed(tasks) — stream results.
• asyncio.TaskGroup() — 3.11+ structured concurrency.
• asyncio.timeout(sec) — 3.11+ deadline.
• asyncio.wait_for(coro, timeout) — older deadline syntax.
• asyncio.create_task(coro) — schedule task, get handle.
• asyncio.Semaphore(N) — concurrency bound.
• asyncio.Queue(maxsize=N) — backpressure.
• asyncio.Lock — mutex.
• asyncio.Event — signal.
• asyncio.to_thread(func, *args) — offload blocking.
• aiohttp.ClientSession — shared HTTP client.
• aiohttp.TCPConnector(limit=100, limit_per_host=10) — pool limits.
• aiohttp.ClientTimeout(total=30, connect=5) — layered timeouts.
• aiolimiter.AsyncLimiter(rate, period) — leaky-bucket rate limit.
• asyncpg.create_pool(dsn, min_size, max_size) — Postgres pool.
• conn.fetch / fetchrow / fetchval — asyncpg reads.
• conn.copy_records_to_table — asyncpg bulk load.
• Never time.sleep in coroutine — use asyncio.sleep.
• Never sync driver on event loop — use async equivalent or to_thread.
• One event loop per process.
• Session / pool per event loop.
• return_exceptions=True on gather for partial-tolerate.
• Retry with backoff — 0.5 * 2 ** attempt.
• raise_for_status — check HTTP status.
• iter_chunked(size) — stream large downloads.
• Sentinel None in Queue — signal worker exit.
• try/except CancelledError — cleanup on cancel.
• asyncio.shield — protect critical cleanup.

Frequently asked questions

When should I use asyncio vs threading vs multiprocessing?

Use asyncio for I/O-bound workloads with high concurrency (100-10K coroutines) — HTTP scraping, DB fan-out, webhook processing. Coroutines are cheap (bytes each), no GIL contention on I/O waits. Use threading for I/O-bound with libraries that lack async support (10-100 threads). Use multiprocessing for CPU-bound work — asyncio and threading can't parallelize CPU because of the GIL; multiprocessing forks processes with independent GILs. Rule: async first for new I/O code; threading for legacy sync libraries; multiprocessing for CPU-heavy compute like NumPy / Pandas without release-GIL fast paths.

What's the difference between gather and as_completed?

gather awaits all tasks and returns results in submission order — good when you need everything and the collection is bounded. as_completed yields tasks as they finish (order not preserved), letting you start processing responses immediately without waiting for the slowest — good for streaming large batches. Since Python 3.11, TaskGroup provides structured concurrency — automatic sibling cancellation on error, cleaner exception handling via ExceptionGroup. Rule: TaskGroup for fail-fast; gather with return_exceptions for tolerate-partial; as_completed for stream-process.

How do I handle a failed task without cancelling siblings?

asyncio.gather(*tasks, return_exceptions=True) — instead of raising the first exception, collects them as results (exceptions and normal returns interleaved). Iterate the results and check isinstance(r, Exception). Alternative with TaskGroup: catch ExceptionGroup and handle per-exception with except* syntax (3.11+). Rule: return_exceptions=True when partial success is acceptable; TaskGroup when any failure should cancel everything.

How do I rate-limit an async HTTP scraper?

Use aiolimiter.AsyncLimiter(max_rate, time_period) in an async with block around each HTTP call. It's a leaky-bucket rate limiter — up to max_rate immediate requests, then paced. For per-host rate limiting, combine with aiohttp.TCPConnector(limit_per_host=N). For hierarchical limits (global + per-endpoint), nest two limiters. Rule: rate limit at the HTTP call site, not at the queue level; connector handles per-host concurrency automatically.

Why is asyncpg faster than psycopg async?

asyncpg implements the PostgreSQL wire protocol natively in Python + C, optimized for the event loop from the ground up. psycopg wraps libpq (C) with an async interface, which introduces overhead at the interop boundary and doesn't take full advantage of asyncio primitives. Typical benchmarks show asyncpg 3-10× faster on DE workloads (bulk reads, small writes, prepared statements). psycopg3 has narrowed the gap significantly but asyncpg still wins for hot paths. Rule: asyncpg for hot Postgres async paths; psycopg3 async for compatibility with sync code sharing.

What's the one-loop-per-process rule?

An event loop is bound to the thread that started it. Sharing a loop across threads or processes leads to race conditions and undefined behavior. Sessions and pools (aiohttp.ClientSession, asyncpg.Pool) are bound to their loop. Never share them across forks. If you use multiprocessing, each child creates its own loop and session/pool in if __name__ == "__main__". Rule: one loop per process; sessions/pools follow the loop; use asyncio.run() at the top of each process's entry point.

Practice on PipeCode

• Drill the SQL practice library → — 450+ DE-focused questions covering async patterns.
• Sharpen SQL optimization drills → for tuning async DB queries.
• Layer SQL join drills → — async fan-out across joined tables.
• Warm up with aggregation drills → for async aggregation over multiple sources.
• For the broader SQL interview surface, take the SQL for Data Engineering course →.

Pipecode.ai is Leetcode for Data Engineering — every `asyncio pipelines` pattern above ships with hands-on practice rooms where you compose `aiohttp.ClientSession` with `TCPConnector(limit_per_host=10)` for polite scraping, wire `asyncpg` pools with prepared statements and `copy_records_to_table` for bulk Postgres load, layer `AsyncLimiter` + `Semaphore` + `Queue` for the three-way concurrency control, switch between `gather`, `as_completed`, and 3.11+ `TaskGroup` for the right structured-concurrency shape, and finally offload sync SDKs via `asyncio.to_thread` without blocking the loop — the exact async fluency that senior DE and backend interviews probe.

Practice SQL now →
Optimization drills →