python 36 lines · 8 steps

Bounded-concurrency HTTP fetching with asyncio

Fetch many URLs in parallel while capping in-flight requests with a semaphore, returning a structured result for each.

Explained by highlit

Intermediate async concurrency semaphore http error-handling

1import asyncio
2from dataclasses import dataclass
3 
4import aiohttp
5 
6 
7@dataclass
8class FetchResult:
9    url: str
10    status: int | None
11    body: str | None
12    error: str | None = None
13 
14 
15async def fetch_one(
16    session: aiohttp.ClientSession,
17    semaphore: asyncio.Semaphore,
18    url: str,
19    timeout: float = 10.0,
20) -> FetchResult:
21    async with semaphore:
22        try:
23            async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
24                return FetchResult(url=url, status=resp.status, body=await resp.text())
25        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
26            return FetchResult(url=url, status=None, body=None, error=str(exc))
27 
28 
29async def fetch_all(urls: list[str], max_concurrency: int = 8) -> list[FetchResult]:
30    semaphore = asyncio.Semaphore(max_concurrency)
31    async with aiohttp.ClientSession() as session:
32        tasks = [
33            asyncio.create_task(fetch_one(session, semaphore, url))
34            for url in urls
35        ]
36        return await asyncio.gather(*tasks)

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1A shared asyncio.Semaphore caps how many coroutines run their critical section at once, throttling concurrency without blocking the loop.
2Returning a result object per URL instead of raising lets partial failures coexist with successes in a single gathered list.
3Reusing one ClientSession across all requests amortizes connection pooling and TLS setup for every fetch.

Related explainers

rust

use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json

Intermediate 6 steps

ruby

require "shellwords"
require "open3"
 
module Backup

Building safe shell commands in Ruby

shell-out subprocess command-injection

Intermediate 7 steps

python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()

Building a WebSocket chat with FastAPI

websockets broadcast connection-management

Intermediate 9 steps

python

import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy

Intermediate 7 steps

rust

use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel

Advanced 7 steps

javascript

import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks

Advanced 10 steps

Share this explainer

Here's the card — post it anywhere.

Bounded-concurrency HTTP fetching with asyncio — share card

Made with highlit — turn any snippet into a walkthrough like this in about a minute.

Explain your code