3. From requests to httpx.AsyncClient

httpx.AsyncClient mirrors the requests API shape almost beat for beat: same parameter names, same raise_for_status(), same response object. The delta is four lines of plumbing (an async on the function, an await on the call, an async with for the client, and an asyncio.run entry point) and one mental-model shift: from "the call blocks" to "the call yields." The shift is what unlocks the concurrency in Section 4, but it is also what makes the canonical async footgun a quiet one. A sync call inside an async def blocks the event loop and silently defeats the concurrency. This page sets the swap up and then proves the footgun with a wall-clock demo.

From requests to httpx: the async HTTP client

The requests library you have been using throughout the book is synchronous: each call blocks until the HTTP request completes. The async equivalent is httpx, whose API was designed deliberately to mirror requests. Same parameter names, same response.raise_for_status() (the same method from Chapter 4 with the same semantics), same JSON helper. The library exposes both a sync httpx.Client and an async httpx.AsyncClient; this chapter uses the async one.

Install httpx with:

pip install httpx

Here is a side by side comparison of synchronous requests and async httpx:

import requests

def fetch_weather(city):
    response = requests.get(
        f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=YOUR_KEY"
    )
    response.raise_for_status()
    return response.json()

weather = fetch_weather("London")
print(weather)

import asyncio
import httpx

async def fetch_weather(city):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=YOUR_KEY"
        )
        response.raise_for_status()
        return response.json()

async def main():
    weather = await fetch_weather("London")
    print(weather)

if __name__ == "__main__":
    asyncio.run(main())

The changes are small but important:

• Change def to async def.
• Use httpx.AsyncClient() instead of requests.
• Add await before the client.get() call.
• Use async with to manage the client as an async context manager.
• Wrap the code in a main() coroutine and run it with asyncio.run().

Why async with instead of plain with

Plain with blocks cannot await during cleanup. Closing an HTTP connection means waiting for pending data to flush, which is itself an async operation; async with exists so Python can run async code inside __aenter__ and __aexit__. The alternative is manual await client.aclose() calls in your own try/finally -- workable, but the kind of plumbing every async HTTP client used to make you write by hand. httpx.AsyncClient handles the lifecycle automatically, including on exceptions.

This single-request async version does not look much different from the sync one, and it is not faster either: one request, one wait, one answer. The win shows up in Section 4, where the same AsyncClient drives many concurrent calls and the wall-clock collapses to the slowest one. Before that, two more swap patterns and the blocking-in-async trap that defeats them both.

Converting functions: def → async def

The first step in converting synchronous code to async is changing function definitions. Any function that calls an async operation must itself be async. This propagates upward through your call stack: if A calls B and B is async, then A must also be async.

The two listings below are reference shapes -- we are reading them rather than running them. The async version uses aiofiles for the file-I/O step, which is a third-party package (pip install aiofiles) that ships an async-aware open; we do not exercise it here, but it appears in the listing so the propagation rule becomes concrete. Compare the two side by side and notice how the only functions that needed the async treatment are the ones that touch I/O.

import json

import requests


def fetch_api_data(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()


def process_data(data):
    # Some CPU work: filtering, transforming
    return [item for item in data if item["active"]]


def save_to_file(data, filename):
    with open(filename, "w") as f:
        json.dump(data, f)


def run_pipeline(url, filename):
    raw_data = fetch_api_data(url)
    processed = process_data(raw_data)
    save_to_file(processed, filename)
    return len(processed)

To convert this to async, you make the I/O functions async and add await where they are called:

import asyncio
import httpx
import aiofiles  # async file I/O
import json

async def fetch_api_data(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

def process_data(data):
    # CPU work stays synchronous (no I/O to await)
    return [item for item in data if item["active"]]

async def save_to_file(data, filename):
    async with aiofiles.open(filename, "w") as f:
        await f.write(json.dumps(data))

async def run_pipeline(url, filename):
    raw_data = await fetch_api_data(url)
    processed = process_data(raw_data)
    await save_to_file(processed, filename)
    return len(processed)

Notice that process_data() remains a regular function because it does not do any I/O. Only functions that perform async operations need to be async. Pure computation stays synchronous.

Managing async context managers and sessions

When you use httpx.AsyncClient(), you should manage it with an async context manager (async with). This ensures that connections are properly opened and closed, even if an error occurs.

If you are making many requests, it is more efficient to reuse a single client session rather than creating a new one for each request:

import asyncio
import httpx

async def fetch_url(client, url):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

async def main():
    urls = [
        "https://api.example.com/endpoint1",
        "https://api.example.com/endpoint2",
        "https://api.example.com/endpoint3",
    ]
    
    # Create one client and reuse it for all requests
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [fetch_url(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    return results

if __name__ == "__main__":
    asyncio.run(main())

By creating the client once and passing it to each function, you avoid the overhead of creating and tearing down connections repeatedly. This pattern is especially important when making hundreds or thousands of requests.

The blocking-in-async trap

The most common async footgun is a sync I/O call sitting inside an async def. The function still type-checks. The awaits still look right. The wall-clock just silently collapses back to sequential, because the sync call blocks the event loop while the others are waiting their turn. Here is the proof:

Save this as blocking_trap.py. It runs five "concurrent" fetches in two flavours -- one with requests.get() (sync), one with httpx.AsyncClient (async) -- and prints the wall-clock of each:

import asyncio
import time
import httpx
import requests

URL = "https://httpbin.org/delay/1"  # deliberate 1-second response


async def broken_fetch():
    # WRONG: sync requests.get() inside async def blocks the event loop
    response = requests.get(URL, timeout=5)
    return len(response.text)


async def correct_fetch(client):
    response = await client.get(URL, timeout=5)
    return len(response.text)


async def main():
    # Run five "concurrent" fetches that are secretly sync
    start = time.perf_counter()
    await asyncio.gather(*[broken_fetch() for _ in range(5)])
    broken_elapsed = time.perf_counter() - start

    # Run five concurrent fetches that actually concur
    async with httpx.AsyncClient() as client:
        start = time.perf_counter()
        await asyncio.gather(*[correct_fetch(client) for _ in range(5)])
        correct_elapsed = time.perf_counter() - start

    print(f"Five 'concurrent' calls with requests.get(): {broken_elapsed:.2f}s")
    print(f"Five concurrent calls with httpx.AsyncClient: {correct_elapsed:.2f}s")


if __name__ == "__main__":
    asyncio.run(main())

Run it from the project root:

python blocking_trap.py

You will see roughly:

Five 'concurrent' calls with requests.get(): 5.12s
Five concurrent calls with httpx.AsyncClient: 1.08s

Both functions are async def. Both got awaited via asyncio.gather. One is concurrent; one is sequential pretending to be concurrent. The cost ratio is the cost of getting the swap wrong: 5x slowdown from a single line, with no exception to tell you.

The same trap appears everywhere sync I/O lives: time.sleep() (use asyncio.sleep()), open(...).read() for large files (use aiofiles if the size matters), blocking SQLite calls (the chapter's keystone keeps these sync deliberately and names that decision in Section 6). When you mix sync I/O with async at the function-call level, the event loop has nowhere to run; every "concurrent" task waits for the blocking one. The fix is always the same: use the async-aware library for the I/O you are doing, or hand the sync work to asyncio.to_thread() and await the wrapper.

Common async pitfalls beyond the blocking trap

The trap above is the one that hurts most because it is silent. The other recurring failures are noisier but still worth naming:

• Forgetting await entirely. The Section 2 footgun: fetch_data() returns a coroutine object, your function moves on, the work never runs. python -W error::RuntimeWarning turns this into a hard error during development.
• Not handling exceptions in gather. By default a single task failure raises immediately to the caller -- but the surviving tasks are not cancelled; they keep running in the background and any further exceptions they raise are silently discarded. This is the gotcha TaskGroup was introduced to fix (TaskGroup does cancel siblings). return_exceptions=True is the third shape: nothing raises, and every position in the result list is either a value or an exception object. Section 5 covers when each is the right call.
• Mixing sync and async at the call boundary. You cannot just call an async function from a sync one without an event loop. Either go async all the way to asyncio.run() at the entry point, or wrap individual sync calls with asyncio.to_thread() when you are bridging.

Keep the async surface small, verify each function works in isolation, and let mypy --strict catch the missing awaits statically. The next section is where the concurrency you have set up actually starts paying off: five concurrent fetches in 1 second instead of 5.