Using Python asyncio with the InfluxDB client v2 for batch tasks
You need to push a backlog of millions of telemetry points into InfluxDB — a migration between buckets, a nightly re-aggregation, or a catch-up after a gateway outage — and the synchronous client turns it into an hours-long serial crawl. Each write() blocks the worker for a full HTTP round-trip, so throughput is capped at one request in flight no matter how many CPU cores sit idle. Naively “fixing” this by firing every chunk at once is worse: an unbounded fan-out exhausts the connection pool, floods /api/v2/write, and earns a wall of HTTP 429 Too Many Requests before the job dies with half its data written. This page shows the controlled-concurrency pattern that lands the whole backlog fast and safely — InfluxDBClientAsync with bounded timeouts, a semaphore that caps in-flight writes, chunked Line Protocol, and jittered retry that tells a transient 503 apart from a fatal 401. It is the concurrency-focused companion to the broader Python client orchestration patterns that provision and drive InfluxDB tasks from code.
Prerequisites
Why async, not more threads
The synchronous client blocks the interpreter thread for the entire duration of every HTTP write, so its throughput ceiling is one request in flight per thread. Batch work is almost entirely network-wait, not CPU, which is exactly the workload asyncio is built for: InfluxDBClientAsync yields control back to the event loop while a write is on the wire, letting dozens of writes overlap on a single thread without the memory and context-switch overhead of a thread-per-request pool. The catch is that concurrency has to be bounded. An event loop will happily schedule ten thousand simultaneous coroutines, and InfluxDB will answer that flood with 429s and connection resets. The whole pattern below is really about one thing — extracting maximum overlap while holding in-flight requests at a number the database is happy to serve.
Solution walkthrough
1. Open an async client with bounded timeouts and gzip
Instantiate InfluxDBClientAsync inside an async with block so the underlying aiohttp session is always closed, even on exception. Set an explicit timeout so a single stalled request — common during an InfluxDB compaction pause or index rebuild — becomes a fast, retryable error instead of a coroutine that hangs forever and pins a slot in your concurrency budget. Enable enable_gzip so large Line Protocol payloads are compressed before transmission; on high-cardinality telemetry this typically cuts egress by 60–80%.
import logging
from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("influx_async_batch")
def make_async_client(url: str, token: str, org: str) -> InfluxDBClientAsync:
return InfluxDBClientAsync(
url=url,
token=token,
org=org,
timeout=30_000, # ms — hard cap on any single HTTP round-trip
enable_gzip=True, # compress Line Protocol before the wire
)
timeout is in milliseconds here and is the parameter that most often gets left at its default; bounding it is what keeps a wedged connection from silently consuming one of your semaphore permits for the whole run.
2. Chunk the batch and cap concurrency with a semaphore
Two independent limits protect the pipeline. Chunk size caps how many points ride in a single HTTP body — keep it in the 3,000–10,000 range so each request is large enough to amortise HTTP overhead but small enough to avoid oversized payloads and the latency spikes that hit the slowest requests. A semaphore caps how many of those chunk-writes are on the wire at once, which is the real throttle against 429s and pool starvation. The two multiply: chunk_size × max_concurrency is roughly your peak in-flight point count, so size them together.
import asyncio
from typing import Any, Iterator
from influxdb_client import WritePrecision
from influxdb_client.client.exceptions import InfluxDBError
def chunked(records: list[dict[str, Any]], size: int = 5_000) -> Iterator[list]:
"""Split the backlog into fixed-size chunks to bound each HTTP body."""
for i in range(0, len(records), size):
yield records[i:i + size]
async def write_chunk(
write_api,
bucket: str,
chunk: list[dict[str, Any]],
sem: asyncio.Semaphore,
) -> None:
async with sem: # acquire a permit; blocks once max in-flight is reached
await write_api.write(
bucket=bucket,
record=chunk,
write_precision=WritePrecision.MS,
)
The async with sem line is the concurrency governor: with a semaphore initialised to, say, 8, only eight write_chunk coroutines can be past that line simultaneously; the ninth awaits a released permit. That single object is what turns an uncontrolled flood into a steady, sustainable stream.
3. Wrap each write in jittered exponential backoff
Transient failures are guaranteed at scale — a compaction pause returns 503, a rate-limit trip returns 429, a brief network blip raises a connection error. Retrying is correct, but retrying in lockstep with no jitter synchronises every worker into a thundering herd that prolongs the very outage you are recovering from. Back off exponentially, add randomised jitter to desynchronise the retries, and retry only idempotent writes against retryable status codes — never a 401, which is a credential problem a token rotation must fix, not a transient error.
import random
from functools import wraps
RETRYABLE = {429, 500, 502, 503, 504}
def async_retry(max_retries: int = 5, base: float = 0.5, cap: float = 10.0):
def deco(fn):
@wraps(fn)
async def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return await fn(*args, **kwargs)
except (InfluxDBError, asyncio.TimeoutError, ConnectionError) as e:
code = getattr(e, "response", None) and e.response.status
fatal = code is not None and code not in RETRYABLE
if fatal or attempt == max_retries - 1:
logger.error("write failed (code=%s, fatal=%s): %s",
code, fatal, e)
raise
delay = min(base * (2 ** attempt), cap) * random.uniform(0.5, 1.5)
logger.warning("retry %d/%d in %.2fs (code=%s)",
attempt + 1, max_retries, delay, code)
await asyncio.sleep(delay)
return wrapper
return deco
Applying @async_retry() to write_chunk isolates the retry policy from the write logic. Because deterministic series and timestamps overwrite rather than duplicate in InfluxDB, re-writing a chunk is safe — the operation is idempotent, so a retried write cannot double-count. Honour any Retry-After header the server sends by preferring it over your computed delay when present.
4. Fan out with TaskGroup and drive it from an entrypoint
Compose the pieces: one client, one semaphore, one coroutine per chunk, all supervised by asyncio.TaskGroup. TaskGroup is strictly better than a bare gather here because if any chunk exhausts its retries and raises, the group cancels the siblings and propagates the error rather than letting the process exit 0 with a partially written backlog.
async def run_batch(url, token, org, bucket, records, max_concurrency=8):
sem = asyncio.Semaphore(max_concurrency)
@async_retry()
async def _guarded(write_api, chunk):
await write_chunk(write_api, bucket, chunk, sem)
async with make_async_client(url, token, org) as client:
write_api = client.write_api()
chunks = list(chunked(records, size=5_000))
async with asyncio.TaskGroup() as tg: # Python 3.11+
for chunk in chunks:
tg.create_task(_guarded(write_api, chunk))
logger.info("wrote %d points across %d chunks", len(records), len(chunks))
# asyncio.run(run_batch(URL, TOKEN, ORG, "downsampled_telemetry", records))
On Python 3.9 or 3.10, replace the TaskGroup block with await asyncio.gather(*(_guarded(write_api, c) for c in chunks)) — but pass return_exceptions=False (the default) so a failed chunk still propagates instead of being silently collected into the results list.
Gotchas and edge cases
CPU-bound work on the event loop starves every write. If you build Line Protocol, parse CSV, or serialise a large Pandas frame inside the coroutine, that synchronous CPU work blocks the single event-loop thread — no other write can make progress while it runs, and your carefully tuned concurrency collapses to serial. Do the heavy transformation before the async phase, or push it into a thread with asyncio.to_thread() so the loop stays free to service network I/O.
Concurrency set higher than the connection pool just queues. The semaphore admits N coroutines, but aiohttp’s connector has its own connection limit; if your max_concurrency exceeds it, the extra coroutines block waiting on a socket rather than actually overlapping. Match the semaphore to a pool sized for the same fan-out, and treat sustained 429s as the signal to lower concurrency, not raise it — the database is telling you its ingest ceiling.
gather without propagation hides partial-write failures. asyncio.gather(..., return_exceptions=True) returns exceptions as values, so a batch that failed to write half its chunks can still look successful to the caller. Either use TaskGroup (which cancels and raises) or keep return_exceptions=False, and always log the written-vs-expected point count so a short write is visible.
Verification
After the run, confirm the destination actually received the expected volume rather than trusting the process exit code. Count the points landed in the batch window and compare against len(records):
from(bucket: "downsampled_telemetry")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sensor_readings")
|> count()
|> group()
|> sum()
From the CLI, a fast smoke check that the write path and token are healthy before launching a long backfill:
influx bucket list --org "$INFLUX_ORG"
A returned count matching your input — with no gaps across the batch window — confirms the fan-out completed without a silent short write.
Related
- Python client orchestration patterns — the pooled client, idempotent provisioning, and retry policy this async pattern extends.
- Writing robust Flux scripts for automated data rollups — replay-safe transformation logic the batched points feed into.
- Best practices for bucket partitioning in IoT telemetry — sizing the source and destination buckets a batch job moves data between.
Up one level: Python Client Orchestration Patterns