Working with APIs and JSON using requests

Most of the data a modern business uses does not live in CSV files. It lives behind APIs (application programming interfaces): web addresses you can send a request to and get structured data back. Payment gateways, delivery trackers, CRMs, weather services, government open-data portals and your own company's microservices all speak this language. Being able to call an API from Python is what turns an analyst who waits for exports into one who pulls fresh data on demand.

In this lesson you will use the requests library to call a small FreshBasket API that runs on your own computer (the mock_server.py file in the data folder; start it in a second terminal with python mock_server.py before you run the code, and leave it running). We make GET requests with parameters, loop through paginated results, handle errors and timeouts, retry a flaky service politely, send data with POST, and flatten nested JSON into a DataFrame. These are exactly the patterns you will use against real services; only the base URL and the authentication change.

All examples target http://127.0.0.1:8000, a server on your machine, so there is no internet requirement and no account to create.

Worked examples

Example 1: Your first request

Question: Is the API up, and what does a response look like?

import requests

BASE = "http://127.0.0.1:8000"

resp = requests.get(f"{BASE}/api/status", timeout=5)
print("status code :", resp.status_code)
print("content type:", resp.headers["Content-Type"])
print("json        :", resp.json())
print("ok?         :", resp.ok)

status code : 200
content type: application/json; charset=utf-8
json        : {'status': 'ok', 'service': 'freshbasket-mock', 'version': '1.0'}
ok?         : True

How it works: requests.get(url) sends an HTTP GET request and returns a Response object. The status code says how it went: 200 means success, 404 not found, 500 server error. The headers carry metadata such as the content type, and .json() parses the body from JSON text into Python dictionaries and lists. .ok is True for any status below 400. Always pass a timeout so a stuck server cannot freeze your script forever.

Reading the result: The server answered with status 200 and the JSON {'status': 'ok', 'service': 'freshbasket-mock', 'version': '1.0'}. The content type tells us the body is JSON, which is why .json() works.

Example 2: Query parameters

Question: How do we ask for a filtered list of products?

resp = requests.get(f"{BASE}/api/products", params={"category": "Dairy", "per_page": 5}, timeout=5)
print(resp.url)

data = resp.json()
print({k: v for k, v in data.items() if k != "results"})
for p in data["results"][:3]:
    print(p["product_id"], p["product_name"], p["unit_price"])

http://127.0.0.1:8000/api/products?category=Dairy&per_page=5
{'page': 1, 'per_page': 5, 'total': 7, 'has_next': True}
P009 Full Cream Milk 1L 68.0
P010 Curd 400g 45.0
P011 Paneer 200g 95.0

How it works: Pass a dictionary as params and requests builds the query string (?category=Dairy&per_page=5) and handles special characters such as spaces and ampersands for you. Never build URLs by gluing strings together. The response is a dictionary with metadata (total, has_next) and the actual rows under results, a very common API shape.

Reading the result: The URL sent was http://127.0.0.1:8000/api/products?category=Dairy&per_page=5. There are 7 dairy products in total; this response returned 5 of them, and has_next is True.

Example 3: Pagination

Question: How do we download all products when the API returns them in pages?

import pandas as pd

def fetch_all_products(per_page=20):
    rows, page = [], 1
    while True:
        r = requests.get(f"{BASE}/api/products", params={"page": page, "per_page": per_page}, timeout=5)
        r.raise_for_status()
        payload = r.json()
        rows.extend(payload["results"])
        if not payload["has_next"]:
            break
        page += 1
    return rows, page

rows, pages = fetch_all_products()
catalog = pd.DataFrame(rows)
print("pages fetched:", pages)
print(catalog.shape)
print(catalog["category"].value_counts().to_dict())

pages fetched: 3
(50, 7)
{'Fruits & Veg': 8, 'Staples': 8, 'Snacks': 8, 'Dairy': 7, 'Beverages': 7, 'Personal Care': 6, 'Household': 6}

How it works: APIs rarely return everything at once. They return one page and tell you whether there is another. The loop keeps requesting the next page until has_next is false, and collects each page's rows in a list that becomes a DataFrame. raise_for_status() turns an error response into an exception so a failed page does not silently produce a short dataset.

Reading the result: We fetched 3 pages and 50 products. The API's own total said 50, so nothing was lost; comparing counts like this is a good sanity check.

Example 4: Path parameters and 404s

Question: How do we fetch one product, and what happens if it does not exist?

ok = requests.get(f"{BASE}/api/products/P001", timeout=5)
missing = requests.get(f"{BASE}/api/products/P999", timeout=5)

print(ok.status_code, ok.json()["product_name"])
print(missing.status_code, missing.json())
print(missing.ok)

200 Banana Robusta
404 {'error': 'product P999 not found'}
False

How it works: Identifiers in the URL path (/api/products/P001) select a single resource. A missing resource returns 404 with a JSON error message, and requests does not raise an exception for that by default: you get a normal response whose status_code you must check. Checking .ok or calling raise_for_status() is how you avoid treating an error message as data.

Reading the result: P001 returned 200 with "Banana Robusta". P999 returned 404 and the message "product P999 not found".

Example 5: Handling errors and timeouts

Question: What can go wrong, and how do we cope?

def safe_get(url, **kwargs):
    try:
        r = requests.get(url, timeout=2, **kwargs)
        r.raise_for_status()
        return r.json()
    except requests.exceptions.HTTPError as e:
        return {"error": f"HTTP {e.response.status_code}"}
    except requests.exceptions.ConnectionError:
        return {"error": "could not connect"}
    except requests.exceptions.Timeout:
        return {"error": "timed out"}

print(safe_get(f"{BASE}/api/products/P002")["product_name"])
print(safe_get(f"{BASE}/api/products/P999"))
print(safe_get("http://127.0.0.1:8999/api/status"))

Tomato Hybrid
{'error': 'HTTP 404'}
{'error': 'could not connect'}

How it works: Three failure types need different responses: the server replied with an error code (HTTPError), nothing is listening (ConnectionError) or it took too long (Timeout). All of them inherit from requests.exceptions.RequestException, so one except RequestException catches any. The wrapper returns a small dictionary instead of crashing, so a batch job can log the failure and continue.

Reading the result: The valid product returned its name, the missing one returned {'error': 'HTTP 404'}, and a port with no server returned {'error': 'could not connect'}.

Example 6: Retry with backoff

Question: The /api/flaky endpoint fails two times out of three. How do we get a result anyway?

import time

def get_with_retry(url, attempts=5, base_delay=0.05):
    for attempt in range(1, attempts + 1):
        r = requests.get(url, timeout=5)
        if r.status_code < 500:
            return r, attempt
        delay = base_delay * 2 ** (attempt - 1)      # 0.05, 0.1, 0.2, ...
        print(f"attempt {attempt} failed with {r.status_code}; waiting {delay:.2f}s")
        time.sleep(delay)
    raise RuntimeError(f"gave up after {attempts} attempts")

resp, used = get_with_retry(f"{BASE}/api/flaky")
print(resp.json(), "after", used, "attempts")

attempt 1 failed with 503; waiting 0.05s
attempt 2 failed with 503; waiting 0.10s
{'message': 'finally!', 'attempt': 3} after 3 attempts

How it works: Server errors (status 500 and above) are often temporary: an overloaded service, a deploy in progress. The standard remedy is to retry with exponential backoff: wait a short time, try again, and double the wait each time, so you do not hammer a struggling server. Only retry errors that might go away (5xx, timeouts), never client errors such as 404 or 422, which will fail forever. Real code also adds random jitter and a cap on total waiting.

Reading the result: The service failed 2 times and succeeded on attempt 3 (this endpoint is designed to fail twice and then succeed, every time you run it) with the message "finally!". Waiting 0.15 seconds in total was a small price for a reliable result.

Example 7: POST with a JSON body

Question: How do we create a new order through the API?

new_order = {"customer_id": "C0001", "items": [{"product_id": "P001", "quantity": 2}, {"product_id": "P010", "quantity": 1}]}

created = requests.post(f"{BASE}/api/orders", json=new_order, timeout=5)
print(created.status_code, created.json()["order_id"], created.json()["status"])

bad = requests.post(f"{BASE}/api/orders", json={"items": []}, timeout=5)
print(bad.status_code, bad.json())

201 O06001 Created
422 {'error': 'customer_id and items are required'}

How it works: requests.post(url, json=payload) converts the dictionary to JSON, sets the Content-Type header and sends it in the request body. A successful creation returns 201 Created, usually with the new resource's ID. A request the server understands but rejects, such as a missing required field, returns 422 (or 400) with an explanation. Reading the error body is how you debug what you sent wrong.

Reading the result: The valid order returned 201 and was given the ID O06001 (the server numbers new orders in sequence, so each time you run this you will get the next ID). The invalid one (no customer_id) returned 422: "customer_id and items are required".

Example 8: Sessions, headers and secrets

Question: How do we send the same headers on every call without repeating ourselves?

import os

token = os.environ.get("FRESHBASKET_TOKEN", "demo-token-for-practice")

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {token}", "User-Agent": "skillancy-tutorial/1.0"})

r = session.get(f"{BASE}/api/products/P003", timeout=5)
print(r.status_code, r.json()["product_name"])
print(r.request.headers["Authorization"][:14] + "...")
print(r.request.headers["User-Agent"])

200 Onion
Bearer demo-to...
skillancy-tutorial/1.0

How it works: A Session remembers headers (and cookies) across requests and reuses the underlying connection, which is faster when you make many calls. Real APIs usually need an API key or token in an Authorization header. Never type secrets directly into code you commit or share: read them from an environment variable with os.environ.get, as above. The mock server ignores authentication, but the pattern is exactly what you will use for real services.

Reading the result: The session sent the header Bearer demo-to... and a custom User-Agent with every request; the server answered 200 for "Onion".

Example 9: Nested JSON to a DataFrame

Question: An order contains a list of items. How do we make that tabular?

orders = [requests.get(f"{BASE}/api/orders/{oid}", timeout=5).json() for oid in ["O00001", "O00002", "O00003"]]
print(list(orders[0].keys()))

flat = pd.json_normalize(orders, record_path="items", meta=["order_id", "customer_id", "payment_method"])
flat["line_total"] = flat["quantity"] * flat["unit_price"]
print(flat.head())
print(flat.groupby("order_id")["line_total"].sum().to_dict())

['order_id', 'customer_id', 'store_id', 'order_date', 'delivery_slot', 'payment_method', 'order_status', 'delivery_minutes', 'discount_pct', 'items']
  product_id  quantity  unit_price order_id customer_id payment_method  line_total
0       P046         1       340.0   O00001       C0326           Card       340.0
1       P015         1        55.0   O00001       C0326           Card        55.0
2       P005         2       180.0   O00001       C0326           Card       360.0
3       P035         2        40.0   O00001       C0326           Card        80.0
4       P030         1        56.0   O00002       C0396            UPI        56.0
{'O00001': 835.0, 'O00002': 364.0, 'O00003': 211.0}

How it works: Nested JSON (a list inside a dictionary) does not fit a flat table directly. pd.json_normalize with record_path="items" creates one row per item, and meta copies chosen order-level fields onto each row, so you can compute line totals and aggregate per order. This is the standard way to turn API responses into analysis-ready data.

Reading the result: The three orders flatten into 9 item rows. The order totals computed from the API data are {'O00001': 835.0, 'O00002': 364.0, 'O00003': 211.0}.

Example 10: Reconcile the API with the CSV

Question: Does the API return the same data as our file?

csv_orders = pd.read_csv("orders.csv").set_index("order_id")
csv_items = pd.read_csv("order_items.csv")

oid = "O00001"
api = requests.get(f"{BASE}/api/orders/{oid}", timeout=5).json()
api_total = sum(i["quantity"] * i["unit_price"] for i in api["items"])
csv_total = (csv_items.loc[csv_items["order_id"] == oid, "quantity"] * csv_items.loc[csv_items["order_id"] == oid, "unit_price"]).sum()

print("API status   :", api["order_status"], "| CSV status:", csv_orders.loc[oid, "order_status"])
print("API total    :", api_total, "| CSV total :", csv_total)
print("match:", api_total == csv_total and api["order_status"] == csv_orders.loc[oid, "order_status"])

API status   : Delivered | CSV status: Delivered
API total    : 835.0 | CSV total : 835.0
match: True

How it works: When two sources should agree, test it. Pull one record from each and compare the facts that matter (status, total). Reconciliation catches mapping bugs, stale caches and truncated pages early, before a wrong number reaches a report.

Reading the result: For order O00001, both sources report status "Delivered" and a total of ₹835, so they agree.

Common mistakes

1. No timeout. Without timeout=, a request can hang forever.

2. Not checking the status code, then parsing an error message as if it were data.

3. Ignoring pagination, so you analyse only the first page and never notice.

4. Retrying everything. Retry only temporary failures (5xx, timeouts), not 4xx errors.

5. Hard-coding secrets in scripts that end up in version control.

requests.get(f"{BASE}/api/products/P999", timeout=5).raise_for_status()

Traceback (most recent call last):
  File "example.py", line 1, in <module>
    requests.get(f"{BASE}/api/products/P999", timeout=5).raise_for_status()
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
  File "example.py", line 1167, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 404 Client Error: Not Found for url: http://127.0.0.1:8000/api/products/P999

Practice questions

Practice 1 (Easy): Count products by category via the API

Use the category parameter to count the products in "Staples" without downloading the others.

r = requests.get(f"{BASE}/api/products", params={"category": "Staples", "per_page": 1}, timeout=5)
print(r.json()["total"], len(r.json()["results"]))

8 1

Practice 2 (Medium): Average price by category from the API

Fetch the full catalogue with the pagination helper and compute the mean price per category, comparing with products.csv.

api_mean = catalog.groupby("category")["unit_price"].mean().round(2)
csv_mean = pd.read_csv("products.csv").groupby("category")["unit_price"].mean().round(2)
print(api_mean.to_dict())
print(api_mean.equals(csv_mean))

{'Beverages': 117.14, 'Dairy': 80.14, 'Fruits & Veg': 102.38, 'Household': 165.0, 'Personal Care': 167.83, 'Snacks': 98.25, 'Staples': 183.5}
True

Practice 3 (Hard): A reusable client with retries

Write a function fetch_order(order_id) that returns the order JSON, retries on server errors using get_with_retry, and returns None for a 404. Test it on one good ID and one bad ID.

def fetch_order(order_id):
    r, _ = get_with_retry(f"{BASE}/api/orders/{order_id}")
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()

good = fetch_order("O00010")
bad = fetch_order("O99999")
print(good["order_id"], len(good["items"]), bad)

O00010 5 None

Key takeaways

• An API call is a request (URL, method, params, headers, body) and a response (status, headers, body).
• Use params= for query strings, .json() for the body, and always set a timeout.
• Loop through pages until has_next is false, and verify the total.
• Check .ok or call raise_for_status(); handle HTTP, connection and timeout errors separately.
• Retry only temporary failures, with exponential backoff.
• POST with json=; 201 means created, 4xx means your request was wrong.
• Keep secrets in environment variables, and flatten nested JSON with json_normalize.