Python · API · Advanced
Async Requests with httpx
Use httpx with asyncio to make concurrent API requests and speed up multi-call scripts.
Quick topic start and explanations before exercises (exercises below):
Parallel requests with asyncio.gather
#httpx and asyncio reference
#Exercises:
First async request
#Write an async function using httpx.AsyncClient that fetches /api/products/ and prints the product count. Run it with asyncio.run().
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_products():
# Use httpx.AsyncClient to fetch /api/products/
pass
asyncio.run(fetch_products())
Solution
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_products():
async with httpx.AsyncClient() as client:
response = await client.get(f'{BASE_URL}/api/products/')
data = response.json()
print(data['count'])
asyncio.run(fetch_products())
Async fetch of one endpoint
#Write an async function fetch_products() that uses httpx.AsyncClient to fetch /api/products/ and returns the count. Call it with asyncio.run() and print the result.
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_products():
# Use AsyncClient to fetch products and return count
pass
print(asyncio.run(fetch_products()))
Solution
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_products():
async with httpx.AsyncClient() as client:
response = await client.get(f'{BASE_URL}/api/products/')
return response.json()['count']
print(asyncio.run(fetch_products()))
Fetch multiple products in parallel
#Write a coroutine fetch_product(client, product_id) that fetches /api/products/{id}/. Use asyncio.gather() to fetch products with ids 1 to 5 simultaneously. Print the name and price of each.
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_product(client, product_id):
# GET /api/products/{product_id}/
pass
async def main():
async with httpx.AsyncClient() as client:
# Gather and run 5 requests in parallel
pass
asyncio.run(main())
Solution
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_product(client, product_id):
response = await client.get(f'{BASE_URL}/api/products/{product_id}/')
return response.json()
async def main():
async with httpx.AsyncClient() as client:
products = await asyncio.gather(*[
fetch_product(client, i) for i in range(1, 6)
])
for p in products:
print(p['name'], p['price'])
asyncio.run(main())
Fetch multiple products by id
#Fetch products with ids 1, 2, 3, 4, and 5 concurrently using asyncio.gather. Print the name and price of each.
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_product(client, product_id):
# GET /api/products/{product_id}/ and return the JSON
pass
async def main():
product_ids = [1, 2, 3, 4, 5]
# Fetch all in parallel and print name + price
pass
asyncio.run(main())
Solution
import httpx
import asyncio
BASE_URL = 'https://apilearn.tukas.dev'
async def fetch_product(client, product_id):
r = await client.get(f'{BASE_URL}/api/products/{product_id}/')
return r.json()
async def main():
product_ids = [1, 2, 3, 4, 5]
async with httpx.AsyncClient() as client:
products = await asyncio.gather(
*[fetch_product(client, pid) for pid in product_ids]
)
for p in products:
print(p['name'], p['price'])
asyncio.run(main())
Compare sequential vs parallel
#Fetch 10 pages of products (/api/products/?page=N) both ways: first sequentially in a for loop (measuring time), then concurrently with asyncio.gather (measuring time). Print both elapsed times.
import httpx import asyncio import time BASE_URL = 'https://apilearn.tukas.dev' # Sequential: fetch pages 1-10 one by one # Parallel: fetch pages 1-10 concurrently # Print elapsed time for each approach
Solution
import httpx
import asyncio
import time
BASE_URL = 'https://apilearn.tukas.dev'
pages = list(range(1, 11))
# Sequential
start = time.time()
for page in pages:
httpx.get(f'{BASE_URL}/api/products/', params={'page': page})
print(f'Sequential: {time.time() - start:.2f}s')
# Parallel
async def fetch_page(client, page):
return await client.get(
f'{BASE_URL}/api/products/', params={'page': page},
)
async def run_parallel():
async with httpx.AsyncClient() as client:
await asyncio.gather(*[fetch_page(client, p) for p in pages])
start = time.time()
asyncio.run(run_parallel())
print(f'Parallel: {time.time() - start:.2f}s')