127. Asyncio And Httpx

⚡ Aryan wants to monitor RAM and call the LLM concurrently — while the LLM thinks, the tool keeps polling psutil. His async code blocks the event loop, forgets to await coroutines, and creates a new HTTP client for every single request.

💡 Fun fact: Python’s asyncio was added in Python 3.4 (2014) via PEP 3156, written by Guido van Rossum himself. The async/await syntax came later in Python 3.5 via PEP 492. The httpx library was created in 2019 as a modern alternative to requests with native async support. asyncio.gather() is inspired by Promise.all() in JavaScript — both take a list of async operations and run them all concurrently, returning results in the same order as the inputs regardless of which finishes first.

⚠️ Watch out: Calling an async def function without await does NOT run the function — it returns a coroutine object. resp = client.post(...) gives you <coroutine object AsyncClient.post at 0x...>. Then resp.raise_for_status() raises AttributeError: 'coroutine' object has no attribute 'raise_for_status'. This is one of the most common async bugs — Python will also emit a RuntimeWarning: coroutine 'post' was never awaited warning to help you catch it.

🤔 Think about it: Creating a new AsyncClient per request (async with _http_client_factory() as client) inside a loop means a fresh TCP connection for each request — the SSL handshake alone takes 100–300ms. A shared client reuses connections from its pool. If you’re making 10 concurrent requests to the same host, how many TCP connections do you need with a shared client vs. a new client per request? What does asyncio.gather() actually do differently from for p in prompts: await fetch(...)?

Learning objectives

Use await on every async client method call
Share one AsyncClient across requests for connection pooling
Use asyncio.gather() to run multiple requests concurrently
Call asyncio.run() exactly once as the synchronous entry point
Understand the difference between sequential await and concurrent gather

Key concepts

async/await — cooperative multitasking
httpx.AsyncClient — async-native HTTP client
asyncio.gather(*tasks) — concurrent fan-out
async with — async context manager
asyncio.run() — run async code from sync context

Try it

import asyncio
from unittest.mock import AsyncMock, MagicMock
<h1 id="httpx-asyncclient-is-injected-via-dependency-injection-in-tests">httpx.AsyncClient is injected via dependency injection in tests.</h1>
<h1 id="the-asynchttpclient-class-below-is-our-stand-in-for-httpx-asyncclient">The AsyncHttpClient class below is our stand-in for httpx.AsyncClient.</h1>
<p>class AsyncHttpClient:
"""Stand-in for httpx.AsyncClient — replaced by mock in tests."""
async def <strong>aenter</strong>(self):
return self
async def <strong>aexit</strong>(self, *args):
pass
async def post(self, url, headers=None, json=None, timeout=None):
raise RuntimeError("Real network call — replace with mock in tests")</p>
<h1 id="http-client-factory-is-patched-in-tests-to-inject-the-mock">_http_client_factory is patched in tests to inject the mock</h1>
<p>_http_client_factory = AsyncHttpClient</p>
<p>async def fetch_llm_response(client, prompt: str, api_key: str) -> str:
"""Fetch LLM response using an existing async client."""
# 💡 : hint : -010 xp client.post() is a coroutine — without await it returns a coroutine object, not a response; add await before client.post() to actually execute the HTTP call
# BUG 1: Not awaiting the coroutine — returns a coroutine object, not a response
resp = client.post(
‘https://api.anthropic.com/v1/messages’,
headers={
‘x-api-key’: api_key,
‘anthropic-version’: ‘2023-06-01’,
‘content-type’: ‘application/json’,
},
json={
‘model’: ‘claude-3-haiku-20240307’,
‘max_tokens’: 128,
‘messages’: [{‘role’: ‘user’, ‘content’: prompt}],
},
timeout=30,
)
resp.raise_for_status()
return resp.json()[‘content’][0][‘text’]</p>
<p>async def fetch_all(prompts: list, api_key: str) -> list:
"""Fetch responses for multiple prompts concurrently."""
results = []
# 💡 : hint : -010 xp a for loop with await runs requests one at a time (sequential) — build a list of coroutines first, then pass them all to asyncio.gather(*tasks) to run concurrently
# BUG 2: Sequential loop with await — runs one at a time, not concurrently
# 💡 : hint : -010 xp create one shared client outside the loop — async with _http_client_factory() as client — then pass it to all fetch_llm_response calls; creating a new client per request opens a new TCP connection each time
# BUG 3: Creating a new client per request — wasteful (new TCP connection each time)
for prompt in prompts:
async with _http_client_factory() as client:
result = await fetch_llm_response(client, prompt, api_key)
results.append(result)
return results</p>
<p>def run_concurrent(prompts: list, api_key: str) -> list:
"""Synchronous entry point that runs the async function."""
return asyncio.run(fetch_all(prompts, api_key))

Concept detail

Async HTTP with asyncio + httpx

asyncio is Python’s built-in async I/O framework. httpx is a modern HTTP client with async support.

Basics

import asyncio
import httpx

async def fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)   # await every coroutine
        return resp.text

Concurrent requests with gather()

async def fetch_all(urls: list[str]) -> list[str]:
    async with httpx.AsyncClient() as client:   # one shared client
        tasks = [fetch_one(client, url) for url in urls]
        return await asyncio.gather(*tasks)      # all run concurrently

The difference:

# Sequential (slow) — 3 × 1s = 3s total
for url in urls:
    result = await client.get(url)   # waits for each before starting next

# Concurrent (fast) — max(1s, 1s, 1s) = 1s total
results = await asyncio.gather(*[client.get(u) for u in urls])

Sync entry point

def main():
    results = asyncio.run(fetch_all(urls))   # one run() call at top level

Common mistakes

Mistake	Fix
`resp = client.post(...)`	`resp = await client.post(...)`
New client per request	Share one `async with httpx.AsyncClient()`
`for url in urls: await fetch(url)`	`await asyncio.gather(*tasks)`
`asyncio.run()` inside async function	Just `await` the coroutine

Solution

import asyncio
from unittest.mock import AsyncMock, MagicMock

class AsyncHttpClient:
    """Stand-in for httpx.AsyncClient — replaced by mock in tests."""
    async def __aenter__(self):
        return self
    async def __aexit__(self, *args):
        pass
    async def post(self, url, headers=None, json=None, timeout=None):
        raise RuntimeError("Real network call — replace with mock in tests")

_http_client_factory = AsyncHttpClient

async def fetch_llm_response(client, prompt: str, api_key: str) -> str:
    """Fetch LLM response using an existing async client."""
    # FIX 1: await the coroutine
    resp = await client.post(
        'https://api.anthropic.com/v1/messages',
        headers={
            'x-api-key': api_key,
            'anthropic-version': '2023-06-01',
            'content-type': 'application/json',
        },
        json={
            'model': 'claude-3-haiku-20240307',
            'max_tokens': 128,
            'messages': [{'role': 'user', 'content': prompt}],
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()['content'][0]['text']

async def fetch_all(prompts: list, api_key: str) -> list:
    """Fetch responses for multiple prompts concurrently."""
    # FIX 2+3: One shared client, all requests run concurrently via gather()
    async with _http_client_factory() as client:
        tasks = [fetch_llm_response(client, p, api_key) for p in prompts]
        return list(await asyncio.gather(*tasks))

def run_concurrent(prompts: list, api_key: str) -> list:
    """Synchronous entry point that runs the async function."""
    return asyncio.run(fetch_all(prompts, api_key))

Tests

import asyncio
from unittest.mock import AsyncMock, MagicMock, patch

def _make_async_client_mock(response_text='mocked response'):
    """Build a mock async client that returns a canned response."""
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.raise_for_status.return_value = None
    mock_resp.json.return_value = {'content': [{'text': response_text}]}

    mock_client = AsyncMock()
    mock_client.post = AsyncMock(return_value=mock_resp)
    return mock_client

def _make_client_factory(response_text='mocked response'):
    """Return a factory that creates a context-manager mock client."""
    mock_client = _make_async_client_mock(response_text)
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=mock_client)
    cm.__aexit__ = AsyncMock(return_value=None)
    def factory():
        return cm
    return factory, mock_client

def test_fetch_llm_response_awaits_post():
    """fetch_llm_response must await client.post()."""
    mock_client = _make_async_client_mock('hello')

    async def run():
        return await fetch_llm_response(mock_client, 'test', 'key')

    result = asyncio.run(run())
    assert result == 'hello', (
        f'Got "{result}" — make sure you await client.post(...)'
    )
    mock_client.post.assert_awaited_once()

def test_fetch_all_uses_gather():
    """fetch_all must run requests concurrently via asyncio.gather."""
    called_gather = False
    original_gather = asyncio.gather

    async def patched_gather(*args, **kwargs):
        nonlocal called_gather
        called_gather = True
        return await original_gather(*args, **kwargs)

    factory, _ = _make_client_factory('ok')

    async def run():
        with patch('asyncio.gather', side_effect=patched_gather):
            # temporarily replace the factory
            import builtins
            old_factory = _http_client_factory
            globals()['_http_client_factory'] = factory
            try:
                return await fetch_all(['x', 'y'], 'k')
            finally:
                globals()['_http_client_factory'] = old_factory

    asyncio.run(run())
    assert called_gather, 'fetch_all must use asyncio.gather() for concurrency'

def test_fetch_all_returns_all_responses():
    factory, _ = _make_client_factory('ok')
    old_factory = globals()['_http_client_factory']
    globals()['_http_client_factory'] = factory
    try:
        results = asyncio.run(fetch_all(['p1', 'p2', 'p3'], 'key'))
    finally:
        globals()['_http_client_factory'] = old_factory
    assert len(results) == 3

def test_run_concurrent_returns_list():
    factory, _ = _make_client_factory('result')
    old_factory = globals()['_http_client_factory']
    globals()['_http_client_factory'] = factory
    try:
        results = run_concurrent(['hello'], 'key')
    finally:
        globals()['_http_client_factory'] = old_factory
    assert isinstance(results, list)
    assert len(results) == 1

def test_single_client_used_for_all_requests():
    """All requests must share one client (one context manager entry)."""
    entry_count = [0]
    response_text = 'shared'
    mock_client = _make_async_client_mock(response_text)

    cm = MagicMock()
    async def fake_aenter():
        entry_count[0] += 1
        return mock_client
    cm.__aenter__ = fake_aenter
    cm.__aexit__ = AsyncMock(return_value=None)

    old_factory = globals()['_http_client_factory']
    globals()['_http_client_factory'] = lambda: cm
    try:
        asyncio.run(fetch_all(['a', 'b', 'c'], 'key'))
    finally:
        globals()['_http_client_factory'] = old_factory
    assert entry_count[0] == 1, (
        f'Client context manager entered {entry_count[0]} times — '
        'all requests should share one client'
    )