Working with APIs in Python | Python Course | Dataplexa

Working with APIs in Python

An API — Application Programming Interface — is how software systems talk to each other over the internet. When your Python script fetches weather data, submits a payment, posts to Slack, or reads from a spreadsheet, it is talking to an API. REST APIs are the dominant standard, and Python's requests library makes consuming them straightforward and readable.

This lesson covers HTTP methods, request and response anatomy, authentication, error handling, sessions, pagination, and building a reusable API client class.

HTTP Methods

  • GET — retrieve data. Safe and idempotent — calling it multiple times gives the same result with no side effects.
  • POST — create a new resource. Sending the same request twice creates two resources.
  • PUT — replace an existing resource entirely.
  • PATCH — partially update an existing resource.
  • DELETE — remove a resource.

Making GET Requests

requests.get() sends a GET request and returns a Response object containing the status code, headers, and body — usually JSON.

import requests

# Fetch a post from JSONPlaceholder — a free REST API for testing
url = "https://jsonplaceholder.typicode.com/posts/1"
response = requests.get(url)

print("Status:", response.status_code)             # 200
print("Content-Type:", response.headers["Content-Type"])

data = response.json()                             # parse JSON → dict
print("Title:", data["title"])
print("User ID:", data["userId"])

# Query parameters — appended as ?key=value
response = requests.get(
    "https://jsonplaceholder.typicode.com/posts",
    params={"userId": 1, "_limit": 3}
)
posts = response.json()
print(f"
Fetched {len(posts)} posts:")
for post in posts:
    print(f"  [{post['id']}] {post['title'][:40]}...")

# Response attributes at a glance
print("
Encoding:", response.encoding)
print("Headers keys:", list(response.headers.keys())[:3])
Status: 200 Content-Type: application/json; charset=utf-8 Title: sunt aut facere repellat provident occaecati User ID: 1 Fetched 3 posts: [1] sunt aut facere repellat provident occ... [2] qui est esse... [3] ea molestias quasi exercitationem repel... Encoding: utf-8 Headers keys: ['Date', 'Content-Type', 'Transfer-Encoding']
  • response.status_code — 200 OK, 201 Created, 400 Bad Request, 401 Unauthorized, 404 Not Found, 429 Rate Limited, 500 Server Error.
  • response.json() — parses the JSON body into a Python dict or list.
  • params=requests encodes the dict as a query string automatically.
  • response.text — raw string; response.content — raw bytes (for files and images).

POST, PUT, PATCH, and DELETE

import requests

BASE = "https://jsonplaceholder.typicode.com"

# POST — create a new resource (returns 201)
new_post = {"title": "My New Post", "body": "Hello world", "userId": 1}
r = requests.post(f"{BASE}/posts", json=new_post)
print("POST status:", r.status_code)    # 201 Created
print("Created:", r.json())

# PUT — replace an existing resource entirely
updated = {"title": "Updated Title", "body": "New body", "userId": 1}
r = requests.put(f"{BASE}/posts/1", json=updated)
print("
PUT status:", r.status_code)   # 200 OK
print("All fields:", list(r.json().keys()))

# PATCH — partially update (only specified fields change)
r = requests.patch(f"{BASE}/posts/1", json={"title": "Patched Title"})
print("
PATCH title:", r.json()["title"])

# DELETE — remove a resource (returns 200 with empty body or 204)
r = requests.delete(f"{BASE}/posts/1")
print("DELETE status:", r.status_code)  # 200 OK
POST status: 201 Created: {'title': 'My New Post', 'body': 'Hello world', 'userId': 1, 'id': 101} PUT status: 200 All fields: ['title', 'body', 'userId', 'id'] PATCH title: Patched Title DELETE status: 200
  • Use json= to send JSON — requests serialises and sets Content-Type: application/json automatically.
  • Use data= for form-encoded data; use files= for multipart file uploads.
  • 201 is the correct status for a successful POST; 204 No Content for DELETE with no response body.

Error Handling

Network requests can fail in many ways. Always handle errors explicitly — set a timeout, catch HTTP errors, and handle connection failures.

import requests
from requests.exceptions import RequestException, Timeout, ConnectionError

def safe_get(url, params=None, timeout=5):
    """GET request with full error handling."""
    try:
        response = requests.get(url, params=params, timeout=timeout)
        response.raise_for_status()   # raises HTTPError for 4xx and 5xx
        return response.json()

    except Timeout:
        print(f"Timed out after {timeout}s: {url}")
    except ConnectionError:
        print(f"Could not connect: {url}")
    except requests.exceptions.HTTPError as e:
        code = e.response.status_code
        if   code == 401: print("Unauthorised — check your API key")
        elif code == 403: print("Forbidden — insufficient permissions")
        elif code == 404: print(f"Not found: {url}")
        elif code == 429: print("Rate limit hit — slow down")
        elif code >= 500: print(f"Server error {code}")
        else:             print(f"HTTP error {code}: {e}")
    except RequestException as e:
        print(f"Request error: {e}")
    return None

data = safe_get("https://jsonplaceholder.typicode.com/users/1")
if data:
    print(f"User: {data['name']} — {data['email']}")

safe_get("https://jsonplaceholder.typicode.com/posts/99999")
User: Leanne Graham — Sincere@april.biz Not found: https://jsonplaceholder.typicode.com/posts/99999
  • response.raise_for_status() — raises HTTPError for any 4xx or 5xx status automatically.
  • Always set a timeout — without it, a slow server hangs your script indefinitely.
  • Catch specific exceptions before the general RequestException.

Authentication

import requests, os

# 1. API Key in query string
requests.get("https://api.example.com/data", params={"api_key": "your_key"})

# 2. API Key in header (most common)
requests.get("https://api.example.com/data", headers={"X-API-Key": "your_key"})

# 3. Bearer token (OAuth2 / JWT)
token = "eyJhbGciOiJIUzI1NiIs..."
requests.get("https://api.example.com/protected",
             headers={"Authorization": f"Bearer {token}"})

# 4. Basic Auth — username and password
requests.get("https://api.example.com/resource", auth=("user", "pass"))

# Best practice — load from environment variables, never hardcode
api_key = os.environ.get("MY_API_KEY", "")
headers  = {"X-API-Key": api_key}
print("API key set:", bool(api_key))

# In development: use python-dotenv to load a .env file
# from dotenv import load_dotenv; load_dotenv()
# api_key = os.environ["MY_API_KEY"]
API key set: False

Sessions — Reusing Connections and Headers

A requests.Session persists headers, auth, and cookies across multiple requests, and reuses the underlying TCP connection — faster for repeated calls to the same server.

import requests

BASE = "https://jsonplaceholder.typicode.com"

with requests.Session() as session:
    # Set headers once — applied to every request in this session
    session.headers.update({
        "Authorization": "Bearer demo-token-123",
        "Accept": "application/json",
        "User-Agent": "DataplexaBot/1.0"
    })

    # Reuse connection and headers automatically
    user  = session.get(f"{BASE}/users/1").json()
    posts = session.get(f"{BASE}/posts", params={"userId": 1}).json()
    todos = session.get(f"{BASE}/todos", params={"userId": 1}).json()

print(f"User:  {user['name']}")
print(f"Posts: {len(posts)}")
print(f"Todos: {len(todos)}")
User: Leanne Graham Posts: 10 Todos: 20

Pagination

import requests

def fetch_all(url, params=None, max_pages=20):
    """Fetch all pages of a paginated API."""
    params   = dict(params or {})
    all_data = []
    page     = 1

    while page <= max_pages:
        params["_page"]  = page
        params["_limit"] = 10
        data = requests.get(url, params=params).json()

        if not data:   # empty → no more pages
            break

        all_data.extend(data)
        print(f"Page {page}: +{len(data)} items (total: {len(all_data)})")
        page += 1

    return all_data

posts = fetch_all("https://jsonplaceholder.typicode.com/posts")
print(f"
Total: {len(posts)} posts")

# Cursor-based pagination pattern (used by many modern APIs)
# next_cursor = response.json().get("next_cursor")
# while next_cursor:
#     r = requests.get(url, params={"cursor": next_cursor})
#     next_cursor = r.json().get("next_cursor")
Page 1: +10 items (total: 10) Page 2: +10 items (total: 20) ... Page 10: +10 items (total: 100) Total: 100 posts

Reusable API Client Class

import requests, os
from requests.exceptions import RequestException

class APIClient:
    """Reusable API client — handles auth, session, and errors."""

    def __init__(self, base_url, api_key=None, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout  = timeout
        self.session  = requests.Session()
        self.session.headers.update({"Accept": "application/json"})
        if api_key:
            self.session.headers["X-API-Key"] = api_key

    def get(self, endpoint, **kwargs):
        return self._request("GET", endpoint, **kwargs)

    def post(self, endpoint, **kwargs):
        return self._request("POST", endpoint, **kwargs)

    def patch(self, endpoint, **kwargs):
        return self._request("PATCH", endpoint, **kwargs)

    def delete(self, endpoint, **kwargs):
        return self._request("DELETE", endpoint, **kwargs)

    def _request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            r = self.session.request(method, url, timeout=self.timeout, **kwargs)
            r.raise_for_status()
            return r.json() if r.content else {}
        except RequestException as e:
            print(f"[{method}] {url} — {e}")
            return None

    def close(self):
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

# Usage — works as a context manager too
with APIClient("https://jsonplaceholder.typicode.com") as client:
    user  = client.get("/users/1")
    posts = client.get("/posts", params={"userId": 1, "_limit": 3})
    new   = client.post("/posts", json={"title": "Test", "body": "x", "userId": 1})

    print(f"User:  {user['name']}")
    print(f"Posts: {len(posts)}")
    print(f"New post ID: {new['id']}")
User: Leanne Graham Posts: 3 New post ID: 101

Quick Reference Table

ConceptMethod / ToolPurpose
GET requestrequests.get(url, params={})Retrieve data
POST requestrequests.post(url, json={})Create a resource
PUT / PATCHrequests.put / patch(url, json={})Replace / partially update
Error handlingresponse.raise_for_status()Raise on 4xx / 5xx automatically
Auth — Bearerheaders={"Authorization": "Bearer token"}OAuth2 / JWT authentication
Sessionrequests.Session()Reuse connection and headers
PaginationLoop with page incrementFetch all pages of results

Practice

Which HTTP method is used to create a new resource on a server?



What does response.raise_for_status() do?



Why should you always set a timeout in requests.get()?



What is the advantage of using requests.Session() for multiple requests?



Where should API keys and tokens be stored instead of being hardcoded?



Which requests.get() argument automatically URL-encodes a dict as query parameters?



Quick Quiz

What is the difference between PUT and PATCH?





Which argument auto-encodes a dict as query parameters in requests.get()?





Which HTTP status code indicates a resource was successfully created?





In Bearer token authentication, where is the token placed?





What condition should a pagination loop check to know there are no more pages?





Which argument in requests.post() serialises a dict as JSON and sets the correct Content-Type header?





NEXT UP
Database Programming in Python
Connecting to SQLite and PostgreSQL, running SQL queries, using parameterised statements, and an introduction to SQLAlchemy ORM.