Python’s asyncio ecosystem has matured dramatically since its introduction in Python 3.4. Today, most major infrastructure libraries have async-native counterparts that let you build high-concurrency applications without threads, callbacks, or complex synchronization primitives.
But asyncio’s core library only provides the event loop and basic primitives. For real-world applications, you need async-compatible libraries for DNS resolution, file I/O, database access, and caching. This is where the aio-libs ecosystem comes in.
Library Overview
| Library | GitHub Stars | Last Updated | Purpose | Replaces |
|---|
| aiodns | 594 | Jul 2026 | Async DNS resolution | socket.getaddrinfo() |
| aiofiles | 3,251 | Jul 2026 | Async file operations | open() / os.read() |
| aiomysql | 1,892 | Mar 2026 | Async MySQL driver | mysql-connector-python |
| aiopg | 1,430 | Dec 2025 | Async PostgreSQL driver | psycopg2 |
| aioredis | 2,283 | Feb 2023 | Async Redis client | redis-py |
aiodns: Async DNS Resolution
DNS lookups are one of the most common sources of latency in network applications. A synchronous getaddrinfo() call blocks the entire event loop, preventing other coroutines from making progress. aiodns solves this by wrapping the pycares library (a C-extension binding to c-ares) to perform non-blocking DNS queries.
Installation
Basic Usage
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| import asyncio
import aiodns
async def resolve_domains():
resolver = aiodns.DNSResolver()
# Resolve A records (IPv4)
result = await resolver.query("github.com", "A")
print(f"github.com → {result[0].host}")
# Resolve AAAA records (IPv6)
result = await resolver.query("google.com", "AAAA")
print(f"google.com IPv6 → {result[0].host}")
# Concurrent resolution
domains = ["python.org", "pypi.org", "stackoverflow.com"]
tasks = [resolver.query(d, "A") for d in domains]
results = await asyncio.gather(*tasks, return_exceptions=True)
for domain, result in zip(domains, results):
if isinstance(result, Exception):
print(f"{domain}: FAILED — {result}")
else:
print(f"{domain}: {result[0].host}")
asyncio.run(resolve_domains())
|
Why Use aiodns
In high-throughput async applications (web scrapers, API gateways, microservice mesh), every millisecond counts. Without aiodns, each new TCP connection blocks the event loop during the DNS lookup phase. With aiodns, hundreds of DNS queries can be in-flight simultaneously.
aiofiles: Async File Operations
Python’s built-in open() is synchronous. Reading a large file, even in an async function, blocks the event loop until the entire read operation completes. aiofiles wraps file I/O in a thread pool executor, yielding control back to the event loop during reads and writes.
Installation
Basic Usage
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| import asyncio
import aiofiles
async def process_logs():
# Read a file asynchronously
async with aiofiles.open("server.log", mode="r") as f:
contents = await f.read()
lines = contents.split("\n")
print(f"Read {len(lines)} log lines")
# Write asynchronously
async with aiofiles.open("output.json", mode="w") as f:
import json
data = {"status": "processed", "line_count": len(lines)}
await f.write(json.dumps(data, indent=2))
# Stream processing for large files
async with aiofiles.open("large_data.csv", mode="r") as f:
async for line in f:
# Process each line without blocking
columns = line.strip().split(",")
await process_row(columns)
asyncio.run(process_logs())
|
When aiofiles Matters
- Web frameworks (FastAPI, aiohttp): Serving uploaded files or generating downloadable reports
- Log processors: Streaming through multi-GB log files without pausing event loop
- Data pipelines: Reading configuration files at startup or writing checkpoint data during processing
aiomysql & aiopg: Async Database Drivers
aiomysql and aiopg provide async-native interfaces to MySQL and PostgreSQL, respectively. Both are built on top of PyMySQL and psycopg2’s protocol implementations, reimplemented for asyncio.
aiomysql Setup
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| import asyncio
import aiomysql
async def query_mysql():
pool = await aiomysql.create_pool(
host="localhost",
port=3306,
user="app",
password="secret",
db="production",
minsize=5,
maxsize=20,
)
async with pool.acquire() as conn:
async with conn.cursor() as cur:
# Parameterized query
await cur.execute(
"SELECT id, name, created_at FROM users WHERE active = %s",
(True,)
)
rows = await cur.fetchall()
for row in rows:
print(f"User {row[1]} (ID: {row[0]})")
pool.close()
await pool.wait_closed()
asyncio.run(query_mysql())
|
aiopg Setup
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import asyncio
import aiopg
async def query_postgres():
dsn = "dbname=myapp user=postgres password=secret host=localhost"
async with aiopg.create_pool(dsn, minsize=5, maxsize=20) as pool:
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("""
SELECT u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name
ORDER BY order_count DESC
LIMIT 10
""")
async for row in cur:
print(f"{row[0]}: {row[1]} orders")
asyncio.run(query_postgres())
|
Connection Pooling
Both aiomysql and aiopg include built-in connection pooling — a critical feature for async applications where hundreds of coroutines may need database access simultaneously. The pool manages connection lifecycle, health checks, and automatic reconnection:
1
2
3
4
5
6
7
8
9
10
11
| # Configure pool with health checks
pool = await aiomysql.create_pool(
host="db.internal",
user="app",
password="secret",
db="production",
minsize=10,
maxsize=50,
pool_recycle=3600, # Recycle connections after 1 hour
echo=False, # Disable SQL logging in production
)
|
aioredis: Async Caching and Pub/Sub
aioredis provides async access to Redis for caching, pub/sub messaging, and data structure operations. While the original aioredis project was archived in favor of redis-py’s async support (redis-py 4.2+), it remains widely used in production and serves as the reference implementation for async Redis access patterns.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| import asyncio
import aioredis
async def redis_example():
redis = await aioredis.from_url("redis://localhost:6379/0")
# Cache with TTL
await redis.set("config:version", "3.2.1", ex=3600)
version = await redis.get("config:version")
# Pub/Sub for real-time events
pubsub = redis.pubsub()
await pubsub.subscribe("deployments", "alerts")
# Atomic operations
await redis.incr("page_views:homepage")
await redis.expire("page_views:homepage", 86400)
await redis.close()
asyncio.run(redis_example())
|
Migration Note
For new projects, consider using redis-py 4.2+ with its native async support (redis.asyncio). The API is nearly identical to aioredis, and it’s officially maintained by the Redis team:
1
2
3
4
| from redis.asyncio import Redis
redis = Redis.from_url("redis://localhost:6379/0")
await redis.ping()
|
Building a Complete Async Stack
Combining these libraries creates a fully async application stack:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| import asyncio
import aiodns
import aiofiles
import aiomysql
async def full_pipeline():
# Step 1: DNS resolution
resolver = aiodns.DNSResolver()
# Step 2: Database connection
pool = await aiomysql.create_pool(
host="db.internal", user="app", password="secret", db="analytics"
)
# Step 3: Query + process
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT url, status FROM endpoints WHERE active = 1")
endpoints = await cur.fetchall()
# Step 4: Concurrent resolution + file writing
tasks = [resolver.query(url, "A") for url, _ in endpoints]
results = await asyncio.gather(*tasks, return_exceptions=True)
async with aiofiles.open("dns_report.json", "w") as f:
import json
report = [
{"url": url, "ip": r[0].host if not isinstance(r, Exception) else str(r)}
for (url, _), r in zip(endpoints, results)
]
await f.write(json.dumps(report, indent=2))
pool.close()
await pool.wait_closed()
print(f"Resolved {len(endpoints)} endpoints")
asyncio.run(full_pipeline())
|
For more async Python patterns, see our Python ORM libraries comparison which covers synchronous and async ORM options including Tortoise ORM and SQLAlchemy’s async extension. If you need to squeeze more performance from your async code, our Python profiling tools guide covers tools that work with async call stacks including py-spy’s native async support.
Production Deployment Patterns
Running async Python in production requires more than just writing async def — you need to handle graceful shutdowns, connection draining, and resource cleanup. Here’s a production-ready pattern using all five libraries:
Graceful Shutdown with Signal Handlers
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
| import asyncio
import signal
import aiodns
import aiomysql
class Application:
def __init__(self):
self.resolver = aiodns.DNSResolver()
self.db_pool = None
self._shutdown = asyncio.Event()
async def start(self):
self.db_pool = await aiomysql.create_pool(
host="db.internal", user="app", password="secret",
db="production", minsize=5, maxsize=20
)
print("Application started")
async def shutdown(self):
print("Shutting down...")
self._shutdown.set()
# Drain connections gracefully
self.db_pool.close()
await self.db_pool.wait_closed()
# Cancel pending tasks
tasks = [t for t in asyncio.all_tasks()
if t is not asyncio.current_task()]
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
print("Shutdown complete")
async def run(self):
await self.start()
try:
await self._shutdown.wait()
finally:
await self.shutdown()
app = Application()
def handle_signal():
asyncio.create_task(app.shutdown())
loop = asyncio.new_event_loop()
loop.add_signal_handler(signal.SIGTERM, handle_signal)
loop.add_signal_handler(signal.SIGINT, handle_signal)
loop.run_until_complete(app.run())
|
Connection Pool Sizing
Determining the right pool size for aiomysql and aiopg depends on your workload characteristics. A common formula for read-heavy workloads:
1
| pool_size = (concurrent_requests * avg_query_time_ms) / 1000
|
For example, if your service handles 1,000 concurrent requests with an average query time of 15ms, you need approximately 15 connections in the pool. Add 20% headroom for burst traffic, giving a pool size of 18-20.
Monitoring Async Health
A health check endpoint that verifies connectivity to all async resources provides early warning of infrastructure issues:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| from fastapi import FastAPI
app = FastAPI()
@app.get("/health")
async def health_check():
checks = {}
# Check DNS
try:
resolver = aiodns.DNSResolver()
await resolver.query("google.com", "A")
checks["dns"] = "ok"
except Exception as e:
checks["dns"] = str(e)
# Check database
try:
async with db_pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT 1")
checks["database"] = "ok"
except Exception as e:
checks["database"] = str(e)
healthy = all(v == "ok" for v in checks.values())
return {"status": "healthy" if healthy else "degraded", "checks": checks}
|
Error Handling and Retry Logic
Network-dependent async operations benefit from exponential backoff with jitter. Here’s a wrapper that adds resilience to aiodns queries:
1
2
3
4
5
6
7
8
9
10
11
12
| import random
async def resilient_dns_query(resolver, hostname, record_type, max_retries=3):
for attempt in range(max_retries):
try:
return await resolver.query(hostname, record_type)
except aiodns.error.DNSError as e:
if attempt == max_retries - 1:
raise
delay = (2 ** attempt) + random.uniform(0, 1)
print(f"DNS query failed (attempt {attempt + 1}), retrying in {delay:.1f}s")
await asyncio.sleep(delay)
|
FAQ
Why do I need aiosn when I could just use loop.getaddrinfo()?
loop.getaddrinfo() is indeed async, but it uses a thread pool internally — each call spawns a thread. For applications making thousands of DNS queries per second, the thread pool becomes a bottleneck and a resource drain. aiodns performs true non-blocking DNS queries via c-ares, with zero thread overhead.
Does aiofiles actually provide non-blocking I/O?
Not at the kernel level — aiofiles uses a thread pool executor under the hood because Python doesn’t have native async file I/O APIs (unlike io_uring on Linux). However, it prevents the event loop from blocking, which is the practical goal. For truly non-blocking file I/O, consider using aio-libs/aiofile which wraps libaio.
Can I use aiomysql and SQLAlchemy together?
Yes — SQLAlchemy 1.4+ includes async support via sqlalchemy.ext.asyncio. When using the async engine, you can configure it to use aiomysql as the underlying driver: create_async_engine("mysql+aiomysql://user:pass@host/db").
What’s the difference between aiopg and asyncpg?
asyncpg is a more modern, higher-performance async PostgreSQL driver written entirely in Python with a custom binary protocol implementation. It’s faster than aiopg but has a different API. aiopg uses psycopg2’s protocol and feels more familiar to developers coming from synchronous psycopg2.
Is aioredis still maintained?
The original aio-libs/aioredis was archived in February 2023 and merged into redis-py 4.2+. The redis.asyncio module in redis-py is now the recommended async Redis client. If you’re on an older codebase using aioredis, migration is straightforward — the API is nearly identical.
💰 想测试你的市场判断力？我用 Polymarket 做预测市场交易——这是全球最大的预测市场平台，从大选结果到技术监管时间线，什么都可以押注。和赌博不同，这是真正的信息市场：你懂的信息越多，胜率越高。我靠预测技术相关事件的走向已经赚了不少。用我的邀请链接注册：Polymarket.com