APIs & Async: HTTP Requests and asyncio

APIs & Async: HTTP Requests and asyncio

Installing Dependencies

pip install requests aiohttp

---

Synchronous HTTP with requests

import requests
import json

# GET request — fetch a resource
def get_user(user_id):
    """Fetch a single user from the JSONPlaceholder API."""
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()    # raises HTTPError for 4xx/5xx
        return response.json()
    except requests.exceptions.Timeout:
        print(f"Request timed out for user {user_id}")
        return None
    except requests.exceptions.HTTPError as e:
        print(f"HTTP error {e.response.status_code}: {e}")
        return None
    except requests.exceptions.ConnectionError:
        print("Could not connect to the server")
        return None

user = get_user(1)
if user:
    print(f"Name: {user['name']}")
    print(f"Email: {user['email']}")
    print(f"City: {user['address']['city']}")

---

POST Request with JSON Body

import requests

def create_post(title, body, user_id):
    """Create a new post via POST request."""
    url = "https://jsonplaceholder.typicode.com/posts"
    payload = {
        "title": title,
        "body": body,
        "userId": user_id,
    }
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer your-token-here",
    }

    response = requests.post(url, json=payload, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()

post = create_post("My First Post", "Content here", user_id=1)
print(f"Created post ID: {post['id']}")
print(f"Title: {post['title']}")

---

Understanding the requests Session

import requests

# Use a Session for multiple requests to the same host
# It reuses the TCP connection and applies shared headers/auth

def fetch_all_posts(base_url="https://jsonplaceholder.typicode.com"):
    with requests.Session() as session:
        session.headers.update({
            "Accept": "application/json",
            "User-Agent": "JST-Academy/1.0",
        })

        # Fetch posts
        resp = session.get(f"{base_url}/posts", timeout=10)
        resp.raise_for_status()
        posts = resp.json()

        # Fetch users
        resp = session.get(f"{base_url}/users", timeout=10)
        resp.raise_for_status()
        users = resp.json()

        # Join: attach user info to each post
        user_map = {u["id"]: u["name"] for u in users}
        for post in posts[:5]:
            author = user_map.get(post["userId"], "Unknown")
            print(f"[{author}] {post['title'][:50]}...")

fetch_all_posts()

---

Async Python with asyncio

import asyncio

# async def defines a coroutine — a function that can pause and resume
async def fetch_data(name, delay):
    print(f"Starting {name}...")
    await asyncio.sleep(delay)     # non-blocking pause (simulates I/O)
    print(f"Finished {name} after {delay}s")
    return f"{name}: done"

# Run a single coroutine
async def main():
    result = await fetch_data("task-1", 2)
    print(result)

asyncio.run(main())

# Run multiple coroutines CONCURRENTLY with gather
async def main_concurrent():
    # These run at the same time — total time ≈ 3s (longest), not 2+1+3=6s
    results = await asyncio.gather(
        fetch_data("task-A", 2),
        fetch_data("task-B", 1),
        fetch_data("task-C", 3),
    )
    print("All done:", results)

asyncio.run(main_concurrent())

---

Async HTTP with aiohttp

import asyncio
import aiohttp

async def fetch_user(session, user_id):
    """Fetch a single user asynchronously."""
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.json()

async def fetch_all_users(user_ids):
    """Fetch multiple users concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_user(session, uid) for uid in user_ids]
        users = await asyncio.gather(*tasks, return_exceptions=True)

    results = []
    for uid, user in zip(user_ids, users):
        if isinstance(user, Exception):
            print(f"Error fetching user {uid}: {user}")
        else:
            results.append(user)
    return results

async def main():
    users = await fetch_all_users([1, 2, 3, 4, 5])
    for u in users:
        print(f"{u['name']} — {u['email']}")

asyncio.run(main())

---

Sync vs Async: When to Use Which

The rule: async shines when you're waiting for I/O (network, disk). If you're doing heavy computation, async won't help — it's still single-threaded.