Advanced Async Python: aiohttp, Async Generators, and Semaphores

Why aiohttp Instead of requests?

You know requests, right? It's solid. It's the de facto standard for HTTP in Python. But here's the thing: it's synchronous, each HTTP call blocks until the response arrives. For async work, that kills everything. You create a task, the task makes a request, and suddenly your entire event loop freezes waiting for that response. It defeats the whole purpose of async programming.

aiohttp is built for async Python from the ground up. It's not a wrapper around synchronous code, it's native async all the way down. This matters more than you might think.

When you use aiohttp, several things happen that make your code dramatically faster:

• Non-blocking I/O: Requests don't block the event loop. While waiting for a response, your event loop schedules other tasks.
• Connection pooling: Reuses TCP connections across multiple requests. Creating a new TCP connection is expensive, three-way handshake, SSL negotiation, etc. Pooling amortizes that cost.
• Streaming: Handle large responses without loading them fully into memory. Imagine fetching a 1GB file, with buffering, your RAM explodes. With streaming, you process it chunk-by-chunk.
• WebSocket support: Real-time bidirectional communication. Perfect for chat apps, live feeds, collaborative tools.

Let's start simple and build understanding. This first example is deliberately minimal so you can see the basic structure clearly before we add complexity on top:

import aiohttp
import asyncio
 
async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()
 
async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch_url(session, 'https://httpbin.org/html')
        print(html[:200])
 
asyncio.run(main())

Notice the async with everywhere? That's the async context manager pattern, and it's critical. It ensures resources (connections, sockets) are properly cleaned up even if exceptions happen. When you exit the block, whether normally or via exception, the resource is released. No dangling connections. No resource leaks that accumulate over thousands of requests.

The real magic happens when you scale. You can call fetch_url thousands of times concurrently, and aiohttp will reuse connections intelligently through its internal connection pooling. Try that with requests, create 1,000 threads each calling requests.get(), and you'll run out of file descriptors. Your OS has a limit (usually 1,024 on Linux), and requests will crash the moment you exceed it. aiohttp handles this gracefully because it's async.

Understanding the Session

The ClientSession is your connection manager. Create it once and reuse it for all requests. Don't create a new session per request, that defeats pooling entirely. Sessions are cheap to create but expensive to use effectively only when reused. Think of it like a database connection pool: one pool, many queries.

The async with context manager ensures the session is properly closed, draining any pending connections and releasing resources. This is especially important if you're making thousands of requests, without proper cleanup, your process leaks connections and eventually crashes.

Connection Pooling Strategy

Connection pooling is one of those features that looks like plumbing detail but has massive performance implications. Every new TCP connection incurs overhead: a three-way handshake, TLS negotiation for HTTPS, server-side session setup. For a handful of requests, this cost is invisible. For thousands of requests per minute, it becomes the dominant bottleneck.

aiohttp maintains a pool of open connections per host. When you make a request to a host you've already talked to, it reuses an existing connection from the pool rather than opening a new one. This keeps latency low and throughput high. By default, aiohttp allows up to 100 connections total and 0 connections per host (meaning no explicit per-host limit). You can tune this for your specific workload by passing a TCPConnector with custom settings.

The key insight is that your semaphore limit and your pool size should be coordinated. If you set a semaphore to allow 20 concurrent requests to a single host, but your pool only keeps 10 connections open, you're creating contention, requests wait not because of your rate limit, but because they're waiting to acquire a connection. Set your connector's limit_per_host to match or exceed your intended concurrency level per host. This eliminates a subtle class of performance problems that are difficult to diagnose once your system is in production. Keep your ClientSession alive for the duration of a batch of work, not just for individual requests, and let the pool do its job.

Async Generators: Streaming Data Without Buffering

Regular generators are great for lazily producing values. You've probably used them:

def count_up(n):
    for i in range(n):
        yield i

Each call to next() resumes the generator, runs until the next yield, then suspends. Memory usage is constant, you're producing one value at a time, not building a list.

But what if your data comes from an async operation? What if you're reading from a network stream, and each value requires waiting for I/O? Regular generators can't handle that, they're synchronous. Enter async generators.

An async generator uses async def and yield. You iterate with async for, not regular for. Each iteration is a suspension point, the event loop can handle other tasks while waiting for the next chunk. This is the killer feature, you're not blocking while waiting for data; you're yielding control back to the event loop and resuming when the next piece arrives. Here's what that looks like in practice:

import aiohttp
import asyncio
 
async def fetch_lines(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            async for line in response.content:
                yield line.decode('utf-8')
 
# Usage
async def process_stream():
    async for line in fetch_lines('https://example.com/stream'):
        print(f"Got: {line}")
 
asyncio.run(process_stream())

What's happening here? The fetch_lines generator opens a connection, then yields each line as it arrives. You're not waiting for the entire response, buffering it all in memory. You're processing it as it comes. This is critical for handling multi-gigabyte files or infinite streams. You could be streaming millions of log entries, and your memory usage stays constant because you're processing one at a time.

Why This Matters in Production

Imagine you're aggregating data from a real-time API. The API streams events continuously, thousands per second. With buffering, you'd try to load all events into memory, and you'd run out of RAM. With async generators, you process each event as it arrives, persist it (to a database, file, message queue), and move to the next. Your memory usage is flat regardless of stream volume.

This pattern scales from hobby projects to infrastructure that processes billions of events daily.

Async Comprehensions

Python 3.6+ lets you use comprehension syntax with async generators. These feel natural if you already use list comprehensions, and they reduce the boilerplate of writing explicit async for loops for simple transformations:

# Async list comprehension
lines = [line async for line in fetch_lines(url)]
 
# Async set comprehension
unique_lines = {line async for line in fetch_lines(url)}
 
# Async dict comprehension (manual counter since enumerate() is not async-iterable)
counter = 0
line_lengths = {}
async for line in fetch_lines(url):
    line_lengths[counter] = len(line)
    counter += 1
 
# Async generator expression
gen = (line.upper() async for line in fetch_lines(url))

It looks exactly like regular comprehensions, just with async in front. Under the hood, it's syntactic sugar for async for loops. These are convenient shortcuts for common patterns.

A word of caution: list comprehensions like [line async for line in fetch_lines(url)] wait for the entire stream before returning. If you're streaming millions of items, that defeats the purpose, you're buffering everything. Use the generator expression (line async for ...) to maintain laziness, or use async for loops to process as you go.

Backpressure Patterns

Async generators solve the "producer is faster than memory" problem, but they introduce a subtler one: what happens when your producer generates data faster than your consumer can process it? This is the backpressure problem, and it's a real concern in production streaming systems.

Without backpressure, a fast producer and a slow consumer means data piles up somewhere, either in memory (unbounded queue growth) or in the network stack (connection buffers fill, then TCP slows the sender, then your OS starts dropping packets). Neither is good.

The solution in async Python is to structure your pipeline so the consumer controls the pace. With async generators, this is natural: the producer only runs when the consumer calls async for to request the next item. If the consumer pauses to do expensive processing, the generator pauses too. This cooperative yielding is backpressure built into the language.

You can also implement explicit backpressure with asyncio.Queue. Set a maxsize on the queue. When the queue is full, await queue.put(item) blocks the producer until the consumer calls await queue.get() to free space. This bounds your memory usage and creates clear coordination between producer and consumer stages. For multi-stage pipelines, fetch, transform, persist, queues between each stage let each stage run at its natural pace without one stage starving or overwhelming the others. Design your async pipelines with backpressure in mind from the start, and you'll avoid the class of production incidents where systems slowly run out of memory under load.

Rate Limiting with asyncio.Semaphore

Here's a trap many developers fall into: they fire up 10,000 concurrent tasks, crash their machine, and wonder why. They blame async for being "unreliable" when really, they've overwhelmed their system.

Semaphores are your answer. A semaphore is a synchronization primitive that limits how many tasks can access a resource simultaneously. Think of it like a bouncer at a club: the club fits 50 people. The bouncer lets people in one at a time, up to 50. The 51st person waits outside until someone leaves. The key insight is that you still create all 10,000 tasks upfront, the event loop handles scheduling them, but only the semaphore-permitted number actually run at once:

import asyncio
import aiohttp
 
async def fetch_with_limit(session, semaphore, url):
    async with semaphore:  # Wait until semaphore allows entry
        async with session.get(url) as response:
            return await response.json()
 
async def fetch_many(urls):
    semaphore = asyncio.Semaphore(10)  # Max 10 concurrent requests
 
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, semaphore, url)
            for url in urls
        ]
        return await asyncio.gather(*tasks)
 
urls = [f'https://httpbin.org/delay/1' for _ in range(100)]
results = asyncio.run(fetch_many(urls))

What's happening?

1. We create a semaphore with a count of 10
2. Each async with semaphore acquires a permit before running
3. Once 10 tasks hold permits, the 11th waits until one finishes
4. It's like a bouncer at a club: only 10 people inside at a time

Without the semaphore, all 100 requests fire instantly. Your network card melts. Your DNS resolver cries. Your machine becomes unresponsive. You might even get timeout errors because the kernel can't handle the load. The semaphore keeps it orderly and predictable.

Why Not Just Create Fewer Tasks?

You might think, "Why not just create 10 tasks instead of 100?" Because batch sizes matter. With fixed task creation, you launch 10, wait for all to complete, launch the next 10, etc. With semaphores, you launch all 100 tasks immediately, but only 10 run concurrently. The difference? Efficiency. Task scheduling overhead is amortized over more work. The event loop can swap tasks more intelligently. For CPU-bound coordination, semaphores beat batching.

Async Context Managers: Guaranteed Cleanup

You've seen async with used with session.get(). Let's understand what's really happening under the hood. The protocol is two methods: __aenter__ and __aexit__. These are the async equivalents of __enter__ and __exit__ from synchronous context managers, the difference is that both can await async operations, so you can do real async work during setup and teardown:

class AsyncDatabaseConnection:
    async def __aenter__(self):
        print("Connecting to database...")
        await asyncio.sleep(0.5)  # Simulate connection
        return self
 
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection...")
        await asyncio.sleep(0.2)  # Simulate cleanup
        return False  # Don't suppress exceptions
 
async def use_db():
    async with AsyncDatabaseConnection() as conn:
        print("Using connection...")
        # If exception happens here, __aexit__ still runs
    print("Done")
 
asyncio.run(use_db())

The __aenter__ method runs when entering the async with block. It's where you acquire resources, establish database connections, open files, allocate memory pools. The __aexit__ method runs on exit, whether things went smoothly or crashed. It's where you release resources.

This is crucial: even if an exception occurs, your resources get cleaned up. No dangling connections. No leaked database transactions. No file handles that prevent deletion. This pattern is everywhere in production code because it's essential for reliability.

Output:

Connecting to database...
Using connection...
Closing connection...
Done

The pattern is rock-solid. No matter what happens in the async with block, success, exception, cancellation, the __aexit__ runs.

Structured Concurrency with TaskGroup (Python 3.11+)

If you're on Python 3.11 or later, you have TaskGroup, which is way better than gather(). The fundamental improvement is in how failure is handled: with gather(), if you use return_exceptions=False (the default), the first exception cancels everything and you lose visibility into what else failed. TaskGroup fixes this by collecting all failures and surfacing them together as an ExceptionGroup:

import asyncio
 
async def task1():
    await asyncio.sleep(1)
    return "Task 1 done"
 
async def task2():
    await asyncio.sleep(2)
    raise ValueError("Task 2 failed!")
 
async def task3():
    await asyncio.sleep(0.5)
    return "Task 3 done"
 
async def main_with_taskgroup():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task1())
        t2 = tg.create_task(task2())
        t3 = tg.create_task(task3())
    # When we exit the context, all tasks have finished or failed
    print("All tasks completed (or failed)")
 
asyncio.run(main_with_taskgroup())

TaskGroup has major advantages:

• Cancellation is automatic: If one task fails, all others are cancelled. No orphaned tasks hanging around.
• Exception groups: All failures are collected, not just the first. You see everything that went wrong.
• Cleaner semantics: No need to track a list of tasks manually. The context manager handles it.

TaskGroup was added in Python 3.11 to fix fundamental issues with gather(). It represents a shift toward more ergonomic, reliable concurrent code.

Error Handling in Async

Error handling in async code deserves its own focused discussion because the failure modes are different from synchronous Python. Exceptions can originate in tasks that are running concurrently, and if you're not careful, they either get silently swallowed or propagate in ways that crash unrelated work.

The core principle is: never let exceptions escape from tasks unhandled. Use return_exceptions=True in gather(), or catch exceptions explicitly inside each task. Unhandled exceptions in tasks will print a warning to stderr but won't crash your program, which sounds safe but is actually dangerous, because you lose error information silently. Always log failures explicitly.

When multiple tasks fail, Python 3.11+ packages them into an ExceptionGroup. You can handle this with except* syntax, which lets you filter by exception type and handle each category separately. This is a powerful pattern for distinguishing between transient errors (network timeouts worth retrying) and permanent ones (HTTP 404, authentication failure) within the same concurrent batch:

async def main_with_exception_handling():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task1())
            tg.create_task(task2())  # This will raise
            tg.create_task(task3())
    except ExceptionGroup as eg:
        print(f"Caught {len(eg.exceptions)} exceptions:")
        for exc in eg.exceptions:
            print(f"  - {type(exc).__name__}: {exc}")
 
asyncio.run(main_with_exception_handling())

This is much better than the old way where one exception in gather() would hide the others. With ExceptionGroup, you get visibility into everything that failed, and you can decide how to handle each type of failure.

For Python < 3.11, you're stuck with gather() and manual exception handling. The return_exceptions=True flag is essential here, without it, the first exception cancels the entire batch and you lose results from tasks that succeeded. With it, exceptions are returned as values in the results list, and you iterate to find them:

results = await asyncio.gather(
    task1(), task2(), task3(),
    return_exceptions=True  # Don't raise, return exceptions as results
)
 
for result in results:
    if isinstance(result, Exception):
        print(f"Task failed: {result}")
    else:
        print(f"Task succeeded: {result}")

This works, but it's verbose and error-prone. You have to remember return_exceptions=True, and you manually iterate to find failures. TaskGroup is superior. Whichever approach fits your Python version, the rule is the same: be explicit about what failure means in your concurrent workload, and never assume that because you launched a task it succeeded.

Building a Production Async Web Crawler

Let's tie it all together. Here's a real web crawler that:

• Fetches multiple URLs concurrently
• Respects rate limits with a semaphore
• Retries failed requests with exponential backoff
• Handles redirects and errors gracefully
• Extracts and stores results

This is not a toy. The structure below, retry logic, per-request timeouts, semaphore-gated concurrency, error capture without crashing the batch, is exactly how production systems handle HTTP at scale. Read through the implementation, then we'll break down the individual decisions:

import asyncio
import aiohttp
from aiohttp import ClientError
import time
from typing import List, Dict, Any
 
class AsyncWebCrawler:
    def __init__(self, max_concurrent: int = 10, timeout: int = 10):
        self.max_concurrent = max_concurrent
        self.timeout = timeout
        self.results: List[Dict[str, Any]] = []
 
    async def fetch_with_retry(
        self,
        session: aiohttp.ClientSession,
        semaphore: asyncio.Semaphore,
        url: str,
        max_retries: int = 3
    ) -> Dict[str, Any]:
        """Fetch a URL with exponential backoff retry logic."""
 
        for attempt in range(max_retries):
            try:
                async with semaphore:
                    async with session.get(
                        url,
                        timeout=aiohttp.ClientTimeout(total=self.timeout)
                    ) as response:
                        content = await response.text()
                        return {
                            'url': url,
                            'status': response.status,
                            'content_length': len(content),
                            'success': True,
                            'error': None
                        }
 
            except asyncio.TimeoutError:
                error_msg = "Timeout"
                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)  # Exponential backoff
                    continue
 
            except ClientError as e:
                error_msg = str(e)
                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
 
            except Exception as e:
                error_msg = f"Unexpected: {type(e).__name__}"
                break
 
        # All retries exhausted
        return {
            'url': url,
            'status': None,
            'content_length': 0,
            'success': False,
            'error': error_msg
        }
 
    async def crawl(self, urls: List[str]) -> List[Dict[str, Any]]:
        """Main crawl method."""
        semaphore = asyncio.Semaphore(self.max_concurrent)
 
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.fetch_with_retry(session, semaphore, url)
                for url in urls
            ]
 
            # Gather all tasks, capturing errors without crashing
            results = await asyncio.gather(*tasks, return_exceptions=True)
 
            self.results = [r for r in results if isinstance(r, dict)]
            return self.results
 
    def summary(self) -> Dict[str, Any]:
        """Generate a summary of crawl results."""
        successful = sum(1 for r in self.results if r['success'])
        failed = len(self.results) - successful
        total_bytes = sum(r['content_length'] for r in self.results)
 
        return {
            'total_urls': len(self.results),
            'successful': successful,
            'failed': failed,
            'total_bytes_fetched': total_bytes,
            'success_rate': successful / len(self.results) if self.results else 0
        }
 
# Usage example
async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/status/500',  # Will fail
        'https://httpbin.org/html',
        'https://httpbin.org/json',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/uuid',
    ]
 
    crawler = AsyncWebCrawler(max_concurrent=3)
    start_time = time.time()
 
    results = await crawler.crawl(urls)
    elapsed = time.time() - start_time
 
    summary = crawler.summary()
    print(f"\nCrawl Summary:")
    print(f"  Total URLs: {summary['total_urls']}")
    print(f"  Successful: {summary['successful']}")
    print(f"  Failed: {summary['failed']}")
    print(f"  Total bytes: {summary['total_bytes_fetched']}")
    print(f"  Time elapsed: {elapsed:.2f}s")
    print(f"  Success rate: {summary['success_rate']:.1%}")
 
    # Show failed URLs
    failed_urls = [r for r in results if not r['success']]
    if failed_urls:
        print(f"\nFailed URLs:")
        for result in failed_urls:
            print(f"  - {result['url']}: {result['error']}")
 
if __name__ == '__main__':
    asyncio.run(main())

Let's break down what makes this production-grade:

Semaphore Rate Limiting: The max_concurrent parameter controls how many requests happen simultaneously. Without this, you'd hammer the server and likely trigger rate-limiting or bans.

Retry with Backoff: Failed requests get retried up to 3 times with exponential backoff (1s, 2s, 4s). Transient network hiccups won't sink your crawl. You'll retry quickly at first (maybe it's a blip), then back off to avoid overwhelming a struggling server.

Timeout Handling: Each request has a timeout. No hanging forever on a slow server. After the timeout, the task fails and retries.

Exception Resilience: Individual request failures don't crash the entire crawl. We capture the error and move on. This is critical for web crawling where you might hit 404s, timeouts, server errors, etc.

Summary Reporting: Post-crawl, we get metrics: success rate, total bytes fetched, timing. Critical for monitoring and alerting.

Run this against 100+ URLs, and you'll see it churn through them in controlled batches. The first three requests fire, while others wait in the semaphore queue. As each completes, the queue processes the next URL. Without the semaphore, it would try to fetch everything at once and likely fail spectacularly.

Common Async Mistakes

Even experienced Python developers make these mistakes when moving into async territory. Knowing them upfront saves hours of debugging.

Mistake 1: Creating a new session per request. This is the most common performance mistake. Sessions are expensive to set up because they initialize the connection pool. Creating a new session for every request means you never benefit from connection reuse, TLS session resumption, or any of the pooling infrastructure. Create one ClientSession at the start of your workload and share it across all requests. If you need session-like isolation between logical groups of work, use separate sessions per group, not per request.

Mistake 2: Not using semaphores. Tasks are lightweight, but concurrent resource usage, network sockets, file descriptors, database connections, is not. If you gather() 10,000 tasks without a semaphore, you'll exhaust OS resources before they all complete. Always size your semaphore to match the resource limit you care about: concurrent outbound connections, database pool size, or API rate limit.

Mistake 3: Blocking the event loop with sync code. Calling time.sleep(), requests.get(), or any synchronous I/O from inside a coroutine blocks the entire event loop, every other task freezes until your blocking call returns. Always use asyncio.sleep(), aiohttp, aiofiles, and other async-native libraries. If you must call synchronous code, use loop.run_in_executor() to run it in a thread pool so the event loop stays responsive.

Mistake 4: Forgetting to clean up resources. Always use async with for sessions, connections, and any resource that needs explicit teardown. Relying on garbage collection to clean up async resources is unreliable, the garbage collector doesn't know to await your cleanup coroutines. Missing cleanup leads to connection leaks that accumulate over time and eventually crash your process or exhaust server-side connection limits.

Mistake 5: Ignoring task cancellation. When tasks are cancelled, which happens automatically in TaskGroup on failure, any await point inside them can raise asyncio.CancelledError. If you catch all exceptions with a bare except Exception, you'll swallow cancellations and prevent proper cleanup. Always let CancelledError propagate, or re-raise it after doing your cleanup. This is one of the trickier async gotchas because it only manifests in failure scenarios that are hard to trigger in development.

Key Takeaways

Here's what you now know:

1. aiohttp beats requests for async work, connection pooling, streaming, non-blocking all the way
2. Async generators let you process streaming data without buffering everything in memory
3. Semaphores prevent you from overwhelming your machine with too many concurrent tasks
4. TaskGroup (Python 3.11+) is cleaner than gather() and handles exceptions properly
5. Async context managers guarantee cleanup even when things fail
6. Real crawlers combine all of this: rate limiting, retries, timeouts, and error resilience

The crawler we built is something you can actually ship. It respects server load, handles failures gracefully, and gives you visibility into what succeeded and what didn't. You can integrate it into a larger pipeline: scrape a site, transform the data, feed it to your ML model, whatever you need.

What's Next?

The patterns in this article, connection pooling, backpressure, semaphore-controlled concurrency, structured error handling, are not advanced tricks. They are the baseline for async Python that behaves reliably in production. Async code that lacks any of these properties will work fine on a laptop under light load and fail in unpredictable ways under real conditions. Start with these patterns from day one, not as retrofits after you've seen your system fail.

From here, the natural next step is understanding how your async code actually performs, which operations are the bottlenecks, where you're spending time, and whether your concurrency settings are optimal. Profiling async Python has its own set of tools and techniques that are different from synchronous profiling. You'll learn to use cProfile, line_profiler, and scalene, interpret their output for concurrent code, and make informed optimization decisions rather than guessing.

The async ecosystem in Python is mature and battle-tested. Tools like aiohttp, asyncio, and async/await syntax are used in production by companies handling millions of requests daily. You're learning not just clever tricks, you're learning infrastructure. The investment pays off every time you need to build something that has to move fast, stay reliable, and scale gracefully.