Async ETL in Python: httpx + asyncio to Pandas DataFrames (2026)
Build async ETL pipelines in Python using httpx and asyncio.TaskGroup with Pydantic V2 validation. Working code, retry policy, rate-limit handling, and the pitfalls that bite in production.
Async ETL in Python pulls data from many REST APIs concurrently using httpx and asyncio, validates each response with Pydantic, and collects the results into a pandas or Polars DataFrame, typically 10 to 50x faster than sequential requests calls. The pattern is a single async client with bounded concurrency, a retry policy, and a streaming collector that hands rows off to a DataFrame at the end. I've shipped this exact pipeline in production at three jobs now, and the parts that bite you are never the async syntax. It's the connection limits, the silent 429s, and the Jupyter event loop.
httpx.AsyncClient with asyncio.TaskGroup is the 2026 default. It replaces the older asyncio.gather pattern and propagates exceptions cleanly.
A bounded asyncio.Semaphore plus httpx.Limits is the only reliable way to avoid hitting per-host connection limits and tripping rate limiters.
httpx and aiohttp are within 10 to 15% of each other on raw throughput in 2026 benchmarks. httpx wins on ergonomics and HTTP/2 support.
Pydantic V2 validation between the HTTP response and the DataFrame catches schema drift weeks earlier than dtype errors in pandas.
In Jupyter, the event loop is already running, so await works at the cell level and asyncio.run() will raise RuntimeError.
Use tenacity with exponential backoff plus jitter for retries. Never retry a POST without an idempotency key.
Why async ETL matters in 2026
So, let's start with the bottleneck. The single most common one I see in Python data pipelines isn't CPU and it isn't pandas. It's a for url in urls: requests.get(url) loop sitting in front of a paginated API. On a vendor endpoint with 80 ms median latency and 4,000 pages, that loop is 320 seconds of pure waiting. The same workload through an httpx.AsyncClient with concurrency=20 finishes in 16 seconds. No new hardware, no thread pool, no Dask cluster. Just I/O multiplexing.
asyncio became the right tool for ETL once three things landed. Python 3.11 added asyncio.TaskGroup, which gives structured concurrency with proper exception propagation (no more "Task exception was never retrieved" warnings drowning your logs). httpx hit 1.0 with stable HTTP/2 multiplexing. And Pydantic V2 became fast enough (5 to 50x its V1 predecessor, per the official Pydantic V2 benchmarks) that you can validate every API response without it becoming the bottleneck. That stack is what I default to in 2026.
Threading still has a place. You'll see concurrent.futures.ThreadPoolExecutor wrapped around requests in legacy code, and it works fine. But it caps out at a few hundred concurrent calls and the failure modes (thread leaks, GIL contention on JSON parsing) are uglier than asyncio's. If you're starting a new pipeline today, start async.
Is httpx faster than aiohttp?
aiohttp is marginally faster on pure throughput, around 10 to 15% in head-to-head benchmarks on synthetic endpoints, but httpx wins on every other axis that matters for ETL: HTTP/2 support, a requests-compatible sync API for tests, proper proxy and trust_env handling, and clean integration with anyio. For a production ETL job pulling from a real API at 80 ms median latency, that 10% throughput gap is invisible. You'll never measure it through the noise of network jitter.
Where the difference actually shows up is in code maintenance. httpx looks like requests: client.get(url, params=...), response.json(), response.raise_for_status(). aiohttp uses async context managers for every response (async with session.get(url) as resp: data = await resp.json()), which is easy to get wrong. Forget the inner await and you get a coroutine where you expected a dict. Honestly, three of the four async ETL bugs I've debugged in the last year were "I forgot to await the response body in aiohttp."
If you already have an aiohttp codebase, don't migrate just for ergonomics. If you're greenfield, pick httpx. Anthropic, OpenAI, and Stripe all ship httpx-based async SDKs now. The ecosystem has converged.
Building an async ETL pipeline with httpx
Here's the minimal pattern. One client, one TaskGroup, structured exception handling, and a list of validated records that you hand to pandas at the end.
import asyncio
import httpx
import pandas as pd
from pydantic import BaseModel, Field
class Order(BaseModel):
id: int
customer_id: int = Field(alias="customerId")
total_cents: int = Field(alias="totalCents")
created_at: str = Field(alias="createdAt")
API = "https://api.example.com/v2/orders"
async def fetch_page(client: httpx.AsyncClient, page: int) -> list[Order]:
r = await client.get(API, params={"page": page, "size": 200})
r.raise_for_status()
return [Order.model_validate(row) for row in r.json()["data"]]
async def run(num_pages: int) -> pd.DataFrame:
limits = httpx.Limits(max_connections=20, max_keepalive_connections=20)
timeout = httpx.Timeout(10.0, connect=5.0)
async with httpx.AsyncClient(limits=limits, timeout=timeout, http2=True) as client:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch_page(client, p)) for p in range(num_pages)]
rows = [order.model_dump() for task in tasks for order in task.result()]
return pd.DataFrame(rows)
df = asyncio.run(run(num_pages=200))
print(df.shape, df.dtypes)
A few things worth flagging. First, the client is created once outside the TaskGroup and reused across every request. This is the single most important thing for performance. A fresh AsyncClient per request kills connection reuse, defeats HTTP/2 multiplexing, and adds 5 to 20 ms of TLS handshake per call. Second, the TaskGroup is doing structured concurrency: if any page raises, all sibling tasks get cancelled cleanly and the exception bubbles out as an ExceptionGroup. No silent task leaks. Third, raise_for_status() happens inside the task so 4xx/5xx codes don't slip into your DataFrame as empty rows.
How do you run async code in Jupyter notebooks?
You don't call asyncio.run() in Jupyter. The kernel is already running an event loop, so it will raise RuntimeError: asyncio.run() cannot be called from a running event loop. Instead, you use top-level await directly in a cell:
That's it. IPython 7+ enables top-level await automatically, and Marimo treats every async cell the same way. If you're in a Python script, an older notebook environment, or any context that complains, the workaround is nest_asyncio:
Avoid nest_asyncio in production code. It monkey-patches the event loop in ways that interact badly with some libraries (notably aiosqlite and certain database drivers) and the behaviour drifts between Python versions. In a notebook for exploration, it's fine. In a deployed pipeline, restructure so asyncio.run is the single entry point. If you're publishing notebooks as reproducible artifacts, the Marimo vs Jupyter comparison walks through the tradeoffs of each runtime's async story in detail.
Controlling concurrency with semaphores and connection limits
The naive "launch 4,000 tasks at once" approach will blow up. You'll exhaust file descriptors, the target API will rate-limit you, and your DNS resolver may give up. There are two layers of control you need, and they do different things.
httpx.Limits caps the underlying connection pool, meaning how many TCP/TLS connections the client is allowed to hold open. Set max_connections roughly equal to your target concurrency. asyncio.Semaphore caps how many tasks can be in flight at once, including time spent parsing JSON and validating with Pydantic. The semaphore is what protects you from queue blow-up when the API gets slow.
SEM = asyncio.Semaphore(20)
async def fetch_page(client: httpx.AsyncClient, page: int) -> list[Order]:
async with SEM:
r = await client.get(API, params={"page": page, "size": 200})
r.raise_for_status()
return [Order.model_validate(row) for row in r.json()["data"]]
With this in place you can tg.create_task 10,000 page fetches and the semaphore admits 20 at a time. Memory stays flat. The official asyncio synchronization primitives docs cover the other shapes (BoundedSemaphore, Lock, Event) but for ETL the plain semaphore is almost always what you want.
Handling 429 rate limiting and retries with tenacity
Every real API will rate-limit you eventually. The right pattern is: retry on transient errors (5xx, 429, network timeouts), back off exponentially with jitter, and surrender after a bounded number of attempts. tenacity is the library I reach for. It's small, dependency-free, and composes with async cleanly.
Exponential backoff with jitter is critical when many of your tasks hit a 429 at the same time. Without jitter, every task waits exactly 1s, then 2s, then 4s, and you get a thundering herd that immediately re-triggers rate limiting. The wait_exponential_jitter helper randomizes within the backoff window. The tenacity documentation covers more advanced policies (per-exception retry, custom wait functions) when you need them.
Validating responses with Pydantic V2 before they hit your DataFrame
Half the data-quality bugs I've debugged would have been caught at the HTTP boundary if someone had run the response through a schema. The pattern that's stuck for me: every API response gets a Pydantic model, every field is typed, and bad rows raise instead of polluting the DataFrame with NaN.
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator
class Order(BaseModel):
id: int
customer_id: int = Field(alias="customerId")
total: Decimal = Field(alias="totalCents")
currency: str = Field(min_length=3, max_length=3)
created_at: datetime = Field(alias="createdAt")
status: str
@field_validator("total", mode="before")
@classmethod
def cents_to_decimal(cls, v: int) -> Decimal:
return Decimal(v) / 100
@field_validator("currency")
@classmethod
def uppercase(cls, v: str) -> str:
return v.upper()
A few things this buys you. datetime parsing happens once at the boundary, so your DataFrame gets a real datetime64[ns] column, not strings you have to pd.to_datetime later and pray no row has a malformed timestamp. The Decimal conversion is exact, which matters for money. And if the upstream API silently renames customerId to customer_uuid next Tuesday, the very first row raises ValidationError instead of silently inserting Nones into a column you'll spend three hours debugging on Thursday. I hit this exact bug shipping a billing reconciliation job in 2024, and Pydantic at the boundary would have saved me an embarrassing standup. For deeper coverage of the DataFrame-side validation story, the Pandera guide for data validation walks through the same checks applied post-load.
From async results to pandas and Polars DataFrames
Once you have a list of validated Pydantic models, getting to a DataFrame is one line. The interesting question is whether you should buffer everything in memory or stream it. For under a million rows it doesn't matter; pd.DataFrame([m.model_dump() for m in models]) is fine. Past that, you want to either write Parquet shards from each page or stream into Polars with a lazy frame.
import polars as pl
import pyarrow as pa
import pyarrow.parquet as pq
async def fetch_and_dump(client, page, out_dir):
orders = await fetch_page(client, page)
table = pa.Table.from_pylist([o.model_dump() for o in orders])
pq.write_table(table, f"{out_dir}/page_{page:05d}.parquet")
# After the TaskGroup finishes:
df = pl.scan_parquet(f"{out_dir}/page_*.parquet").collect()
The streaming approach also gives you cheap resumability: if page 4,317 fails and the job dies, you can restart and skip pages whose Parquet file already exists. I've debated whether to use pandas or Polars for the collection step, and the answer depends on what comes next. If your downstream is more REST APIs and a database insert, pandas is fine. If you're doing aggregations or joins over millions of rows, Polars will pay for itself. The Polars vs pandas 2026 benchmark has the numbers on a real 14 GB dataset. Either way, keep the validation step (Pydantic) outside the DataFrame engine.
For pandas specifically, watch your dtypes. A naive pd.DataFrame.from_records(dicts) will give you object columns for strings and int64 for everything else, which is memory-inefficient at scale. Pass dtype_backend="pyarrow" in pandas 2.2+ to get Arrow-backed columns, or coerce explicitly. The pandas memory optimization guide covers the conversion in more depth.
Common async ETL pitfalls and how I debug them
These are the things that have actually broken pipelines I've shipped, in rough order of how often they bite.
1. Forgetting to share the client. Creating httpx.AsyncClient() per request is the most common performance regression. Symptom: throughput is 3x worse than expected, and packet captures show repeated TLS handshakes. Fix: one client per pipeline run, passed in.
2. Unbounded concurrency. No semaphore, no httpx.Limits. Symptom: 429s from minute one, file descriptor exhaustion, OOM. Fix: asyncio.Semaphore(20) as a default; tune up if and only if you can verify the vendor allows it.
3. Catching the wrong exception. A bare except Exception inside the task swallows CancelledError, which prevents TaskGroup from shutting siblings down. Symptom: pipeline hangs forever on the first failure. Fix: never catch CancelledError; if you must broad-catch, re-raise it explicitly.
4. Mixing sync I/O in. A blocking call to requests, open(), or even a slow JSON deserializer inside an async function blocks the entire event loop. Symptom: concurrency seems capped at 1 regardless of semaphore. Fix: keep async functions purely async, or push blocking work to asyncio.to_thread.
5. Silent partial failures. Using asyncio.gather(*, return_exceptions=True) and never checking the results. Symptom: DataFrame is mysteriously short. Fix: use TaskGroup, which raises an ExceptionGroup on any failure; or explicitly inspect each result.
6. Misreading rate-limit headers. Different APIs use X-RateLimit-Reset (Unix timestamp), Retry-After (seconds), or RateLimit-Reset (seconds-until-reset). They're not interchangeable. Symptom: backoff fires too early and re-triggers 429. Fix: read the API docs, write a small parser per vendor.
If the pipeline lives in production, plug it into a metrics backend: requests-per-second, success/failure split, and p50/p95/p99 latency per endpoint. The production-ready data pipelines guide covers the wider operational picture (monitoring, schema versioning, and reprocessing) that an async ETL job needs to be reliable past day one.
Frequently Asked Questions
How many concurrent requests should I run in async Python ETL?
Start at 10 to 20 concurrent requests. That's enough to saturate most rate-limited APIs without tripping per-host limits. Increase only after you've checked the vendor's documented limit and measured that p95 latency keeps improving. Past 50, you're usually queueing on the server side and not gaining anything.
Is asyncio faster than threading for HTTP requests?
Yes, for I/O-bound HTTP work. asyncio scales to thousands of concurrent requests with one event loop, while threading caps in the low hundreds before context-switch overhead and GIL contention dominate. For mixed CPU/IO workloads (e.g., heavy JSON parsing per response) the gap shrinks; for pure REST pagination, asyncio wins by a wide margin.
Can I use requests with asyncio?
Not directly. requests is synchronous and will block the event loop. You can wrap it in asyncio.to_thread(requests.get, url), which runs it in the default thread pool, but you lose most of the throughput advantage of async. For new code, use httpx; the API is nearly identical to requests.
What's the difference between asyncio.gather and asyncio.TaskGroup?
asyncio.gather runs tasks concurrently and returns results as a list, but its exception handling is awkward. A failure either cancels siblings noisily or (with return_exceptions=True) gets buried in the result list. TaskGroup (Python 3.11+) is structured concurrency: it cancels siblings cleanly on failure and raises an ExceptionGroup you can inspect. Prefer TaskGroup in new code.
How do I avoid memory blow-up when fetching millions of rows asynchronously?
Don't buffer everything in a Python list. Stream each page to a Parquet shard as soon as it arrives, then read the shards with pl.scan_parquet or pyarrow.dataset at the end. Combined with a bounded semaphore, memory stays flat regardless of total row count and the job becomes resumable if it crashes mid-run.
Compare GPTQ, AWQ, bitsandbytes, and GGUF for LLM quantization in Python. Real H100 benchmarks, kernel choices, and a production-ready decision tree for 2026.
A practical 2026 walkthrough of GeoPandas 1.0 for Python geospatial analysis: installing the stack, handling CRS gotchas, running spatial joins, plotting interactive maps, and scaling beyond memory with DuckDB Spatial and GeoParquet.
Compare vLLM, TGI, and SGLang for serving LLMs on your own GPUs in 2026. Throughput numbers, prefix caching, quantization tradeoffs, and a production Docker deployment with monitoring.
