REST API Clients

Reviewed & published by Brayan K

Master building robust API clients, handling authentication, retries, pagination, webhooks, and async operations for production-grade integrations

Part of the free Python course at LearnCodingFast — hands-on lessons with examples you run in your browser, plus practice exercises and a quick quiz.

What You'll Learn

What Is a REST API Client?

A REST API client is code that sends HTTP requests to external services and receives structured responses. Every modern application integrates with external APIs for payments, data, AI, authentication, and more.

ConceptReal-World AnalogyWhat It Does
GET RequestReading a menuFetches data without changing anything
POST RequestPlacing an orderCreates new data on the server
PUT/PATCH RequestModifying your orderUpdates existing data
DELETE RequestCanceling your orderRemoves data from the server
HeadersYour membership cardPasses authentication and metadata

Core Capabilities

Common Use Cases

HTTP Methods Overview

GET - Retrieve Data

Fetch resources without side effects. Safe and idempotent.

POST - Create Resources

Submit data to create new resources.

PUT/PATCH - Update

Modify existing resources. PUT replaces, PATCH updates partially.

The Requests Library

requests is Python's most popular HTTP library. Simple, elegant, and widely used.

Key Features

Building Reusable API Clients

Instead of scattering API calls throughout your codebase, create a dedicated client class that centralizes:

This pattern is used by every major Python SDK (Stripe, OpenAI, AWS, etc.).

🧪 Worked Example — A Client You Can Actually Run

Real HTTP needs a network, and the in-browser editor has none — so this example swaps requests.get for a ten-line stand-in that returns canned data. Every other line is exactly what you would write against a live API: a base URL held in one place, a timeout on the call, raise_for_status() to convert a bad status into an exception, and .json() to get a plain Python dict back.

To point it at a real service, install requests, add import requests, and change one line: fake_get(...) becomes requests.get(...).

# A REST client you can run anywhere — the transport is faked, the shape is real.

class FakeResponse:
    """Stands in for the object requests.get() hands back."""

    def __init__(self, status_code, payload):
        self.status_code = status_code
        self._payload = payload

    def json(self):
        # Real requests parses the JSON body for you and returns a dict.
        return self._payload

    def raise_for_status(self):
        # Real requests raises HTTPError on 4xx/5xx. Skip this call and a 404
        # slides silently past, leaving you parsing an error page as if it were data.
        if self.status_code >= 400:
            raise RuntimeError(f"HTTP {self.status_code} for this request")


FAKE_DB = {
    1: {"id": 1, "name": "Ada Lovelace", "email": "[email protected]"},
    2: {"id": 2, "name": "Grace Hopper", "email": "[email protected]"},
}


def fake_get(url, timeout=5):
    # Pull the trailing id out of ".../users/2" and look it up. A real server
    # does the same job — it just does it over the wire.
    user_id = int(url.rsplit("/", 1)[-1])
    if user_id in FAKE_DB:
        return FakeResponse(200, FAKE_DB[user_id])
    return FakeResponse(404, {"error": "not found"})


class UserClient:
    """The reusable-client pattern: base URL and shared settings live in one place."""

    def __init__(self, base_url, timeout=5):
        self.base_url = base_url.rstrip("/")   # trim a trailing / so joins never double it
        self.timeout = timeout                 # one timeout for every call this client makes

    def get_user(self, user_id):
        url = f"{self.base_url}/users/{user_id}"
        response = fake_get(url, timeout=self.timeout)   # ← requests.get(...) in real life
        response.raise_for_status()                      # 4xx/5xx become exceptions here
        return response.json()                           # a dict, ready to use


client = UserClient("https://api.example.com/")   # note the trailing slash — rstrip handles it

user = client.get_user(1)
print(user["name"])
print(user["email"])

try:
    client.get_user(99)          # nobody has id 99, so the fake server answers 404
except RuntimeError as err:
    print("request failed:", err)

# ✅ Expected output:
# Ada Lovelace
# [email protected]
# request failed: HTTP 404 for this request

The habit to take away: never let a response reach your parsing code without a status check first. Roughly every "why is my JSON decode failing?" bug is a 401 or 404 body being parsed as if it were the data you asked for.

🎯 Your Turn — Finish the Product Client

The fake transport is written for you. Three blanks remain, and all three are the pieces that matter in a real client: check the status, decode the body, report the failure. Run it and compare with the expected output at the bottom.

# 🎯 YOUR TURN — finish the client method (fill in the three ___ blanks)

class FakeResponse:
    def __init__(self, status_code, payload):
        self.status_code = status_code
        self._payload = payload

    def json(self):
        return self._payload

    def raise_for_status(self):
        if self.status_code >= 400:
            raise RuntimeError(f"HTTP {self.status_code}")


PRODUCTS = {10: {"id": 10, "title": "Kettle", "price": 24.99}}


def fake_get(url, timeout=5):
    product_id = int(url.rsplit("/", 1)[-1])
    if product_id in PRODUCTS:
        return FakeResponse(200, PRODUCTS[product_id])
    return FakeResponse(404, {"error": "not found"})


class ProductClient:
    def __init__(self, base_url):
        self.base_url = base_url.rstrip("/")

    def get_product(self, product_id):
        url = f"{self.base_url}/products/{product_id}"
        response = fake_get(url, timeout=5)

        # 👉 call the method that turns a 4xx/5xx response into an exception
        response.___()

        # 👉 call the method that returns the decoded body as a dict
        return response.___()


client = ProductClient("https://api.shop.test")

print(client.get_product(10)["title"])

try:
    client.get_product(999)
except RuntimeError as err:
    # 👉 replace ___ with the text  missing product:  (keep the quotes)
    print("___", err)

# ✅ Expected output:
# Kettle
# missing product: HTTP 404

Error Handling & Retries

Types of Failures

Retry Strategy

Best practices for retries:

Authentication Strategies

Bearer Tokens

Most common for modern APIs:

OAuth 2.0

Standard for user-authorized access:

Async API Clients with aiohttp

For high-performance applications that need to make many concurrent API calls, async clients are essential.

Benefits of Async

Use Cases

Pagination Patterns

Offset-Based Pagination

Simple but can be inefficient for large datasets.

Cursor-Based Pagination

More efficient for large data:

Used by Facebook, Twitter, and modern APIs.

Token-Based Pagination

Returns next page token:

Used by Google APIs, YouTube, Shopify.

Webhooks & Security

What Are Webhooks?

Webhooks are reverse APIs - the external service calls your endpoint when events occur.

HMAC Signature Verification

Always verify webhook signatures to prevent fake requests:

Rate Limiting & Circuit Breakers

Rate Limiting

Prevent exceeding API quotas by implementing client-side rate limiting:

Circuit Breaker Pattern

Prevent cascading failures when external services are down:

Response Validation with Pydantic

External APIs can return inconsistent data. Use Pydantic to validate and transform responses:

This is critical for production systems that depend on external data.

File Operations

Uploading Files

Use multipart/form-data for file uploads. Key considerations:

Downloading Files

NEVER load entire files into memory. Use streaming:

Professional SDK Structure

Well-designed API clients follow this structure:

client/
  __init__.py
  core/
    http_client.py    # Base HTTP operations
    auth.py           # Authentication handlers
    errors.py         # Custom exceptions
    pagination.py     # Pagination helpers
    rate_limit.py     # Rate limiting
  resources/
    users.py          # User endpoints
    orders.py         # Order endpoints
    payments.py       # Payment endpoints
  hooks/
    middleware.py     # Request/response middleware
  utils/
    logging.py        # Logging utilities
    models.py         # Pydantic models
    validators.py     # Custom validators

This separation makes the codebase maintainable, testable, and extensible.

Best Practices Summary

✓ Always Do

✗ Never Do

# REST API Client Examples

# ============================================
# 1. BASIC REQUESTS LIBRARY USAGE
# ============================================

import requests
import time
from typing import Optional, Dict, Any

# Simple GET request
def get_user(user_id: int):
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return response.json()

# POST with JSON body
def create_user(username: str, email: str):
    payload = {"username": username, "email": email}
    response = requests.post(
        "https://api.example.com/users",
        json=payload,
        timeout=5
    )
    return response.json()

# Adding headers and authentication
def authenticated_request(endpoint: str, token: str):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    response = requests.get(
        f"https://api.example.com{endpoint}",
        headers=headers,
        timeout=5
    )
    return response.json()


# ============================================
# 2. REUSABLE API CLIENT CLASS
# ============================================

class APIClient:
    def __init__(self, base_url: str, token: Optional[str] = None):
        self.base_url = base_url
        self.token = token
        self.session = requests.Session()
    
    def _headers(self) -> Dict[str, str]:
        headers = {"Content-Type": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers
    
    def get(self, path: str, **kwargs) -> requests.Response:
        url = self.base_url + path
        return self.session.get(
            url,
            headers=self._headers(),
            timeout=kwargs.get("timeout", 10),
            **kwargs
        )
    
    def post(self, path: str, json=None, **kwargs) -> requests.Response:
        url = self.base_url + path
        return self.session.post(
            url,
            json=json,
            headers=self._headers(),
            timeout=kwargs.get("timeout", 10),
            **kwargs
        )
    
    def put(self, path: str, json=None, **kwargs) -> requests.Response:
        url = self.base_url + path
        return self.session.put(
            url,
            json=json,
            headers=self._headers(),
            timeout=kwargs.get("timeout", 10),
            **kwargs
        )
    
    def delete(self, path: str, **kwargs) -> requests.Response:
        url = self.base_url + path
        return self.session.delete(
            url,
            headers=self._headers(),
            timeout=kwargs.get("timeout", 10),
            **kwargs
        )


# ============================================
# 3. ERROR HANDLING & RETRIES
# ============================================

def safe_api_call(url: str, max_retries: int = 3):
    """Robust API call with error handling and retries"""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            return response.json()
        
        except requests.Timeout:
            print(f"Timeout on attempt {attempt + 1}")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff
        
        except requests.ConnectionError:
            print(f"Connection error on attempt {attempt + 1}")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
        
        except requests.HTTPError as e:
            if e.response.status_code == 429:  # Rate limit
                retry_after = int(e.response.headers.get("Retry-After", 2))
                print(f"Rate limited. Waiting {retry_after}s")
                time.sleep(retry_after)
            elif e.response.status_code >= 500:  # Server error
                print(f"Server error: {e.response.status_code}")
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
            else:
                raise  # Client error - don't retry


# ============================================
# 4. PAGINATION HANDLER
# ============================================

def fetch_all_paginated(base_url: str, limit: int = 100):
    """Handle offset-based pagination"""
    offset = 0
    all_results = []
    
    while True:
        params = {"offset": offset, "limit": limit}
        response = requests.get(base_url, params=params, timeout=10)
        batch = response.json()
        
        if not batch:
            break
        
        all_results.extend(batch)
        offset += limit
        
        # Respect rate limits
        time.sleep(0.1)
    
    return all_results


def fetch_cursor_paginated(base_url: str):
    """Handle cursor-based pagination"""
    results = []
    cursor = None
    
    while True:
        params = {"cursor": cursor} if cursor else {}
        response = requests.get(base_url, params=params, timeout=10)
        data = response.json()
        
        results.extend(data.get("items", []))
        cursor = data.get("next_cursor")
        
        if not cursor:
            break
        
        time.sleep(0.1)
    
    return results


# ============================================
# 5. ASYNC API CLIENT (with aiohttp)
# ============================================

import asyncio
import aiohttp

class AsyncAPIClient:
    def __init__(self, base_url: str, token: Optional[str] = None):
        self.base_url = base_url
        self.token = token
    
    def _headers(self) -> Dict[str, str]:
        headers = {"Content-Type": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers
    
    async def get(self, path: str, session: aiohttp.ClientSession):
        url = self.base_url + path
        async with session.get(url, headers=self._headers()) as response:
            return await response.json()
    
    async def post(self, path: str, json: Dict, session: aiohttp.ClientSession):
        url = self.base_url + path
        async with session.post(url, json=json, headers=self._headers()) as response:
            return await response.json()


async def fetch_multiple_users(user_ids: list):
    """Fetch multiple users concurrently"""
    client = AsyncAPIClient("https://api.example.com")
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            client.get(f"/users/{uid}", session)
            for uid in user_ids
        ]
        return await asyncio.gather(*tasks)


# ============================================
# 6. RESPONSE VALIDATION WITH PYDANTIC
# ============================================

from pydantic import BaseModel, validator

class User(BaseModel):
    id: int
    username: str
    email: str
    active: bool = True
    
    @validator("email")
    def validate_email(cls, v):
        if "@" not in v:
            raise ValueError("Invalid email")
        return v


def get_validated_user(user_id: int) -> User:
    """Fetch and validate user data"""
    response = requests.get(f"https://api.example.com/users/{user_id}")
    data = response.json()
    return User(**data)  # Validates automatically


# ============================================
# 7. RATE LIMITER
# ============================================

class RateLimiter:
    def __init__(self, calls_per_second: float):
        self.rate = calls_per_second
        self.tokens = calls_per_second
        self.last_update = time.time()
    
    def allow(self) -> bool:
        now = time.time()
        elapsed = now - self.last_update
        
        # Add tokens based on elapsed time
        self.tokens = min(
            self.rate,
            self.tokens + elapsed * self.rate
        )
        self.last_update = now
        
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False
    
    def wait_if_needed(self):
        while not self.allow():
            time.sleep(0.1)


# ============================================
# 8. WEBHOOK SIGNATURE VERIFICATION
# ============================================

import hmac
import hashlib

def verify_webhook_signature(
    payload: bytes,
    received_signature: str,
    secret: str
) -> bool:
    """Verify HMAC signature from webhook"""
    expected_signature = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, received_signature)


# ============================================
# 9. FILE UPLOAD & DOWNLOAD
# ============================================

def upload_file(filepath: str, upload_url: str):
    """Upload file to API"""
    with open(filepath, "rb") as f:
        files = {"file": f}
        response = requests.post(upload_url, files=files, timeout=30)
        return response.json()


def download_file(url: str, output_path: str):
    """Download file efficiently with streaming"""
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        with open(output_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)


# ============================================
# 10. CIRCUIT BREAKER PATTERN
# ============================================

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half_open
    
    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half_open"
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failures = 0
        self.state = "closed"
    
    def on_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        
        if self.failures >= self.failure_threshold:
            self.state = "open"


# Usage example
print("API Client examples ready!")
print("\nTry creating an APIClient:")
print("client = APIClient('https://api.example.com', 'your-token')")
print("response = client.get('/users/1')")

Key Takeaways

🎯 Mini-Challenge: Retry With Exponential Backoff

A flaky endpoint fails twice with a 503, then works. Write the retry loop that survives it. Only an outline is given — no logic. Real backoff waits 1s, 2s, 4s; this version uses 0.01s, 0.02s, 0.04s so you are not sat watching a blank screen.

# 🎯 MINI-CHALLENGE: retry with exponential backoff
#
# Copy these six lines exactly as they are — they are your flaky "server":
#
#   import time
#   attempts = 0
#   def flaky():
#       global attempts
#       attempts += 1
#       if attempts < 3:
#           raise RuntimeError("HTTP 503")
#       return {"ok": True}
#
# Now write it:
# 1. def fetch_with_retry(max_attempts=5):
# 2.    loop:  for attempt in range(max_attempts)
# 3.       try:     return flaky()
# 4.       except RuntimeError as err:
# 5.          wait = 0.01 * 2 ** attempt        # 0.01, then 0.02, then 0.04 ...
# 6.          print(f"attempt {attempt + 1} failed ({err}) — retrying in {wait}s")
# 7.          time.sleep(wait)
# 8.    after the loop ends:  raise RuntimeError("gave up after 5 attempts")
# 9. print(fetch_with_retry())
#
# ✅ Expected output:
# attempt 1 failed (HTTP 503) — retrying in 0.01s
# attempt 2 failed (HTTP 503) — retrying in 0.02s
# {'ok': True}

# your code here

Two traps. The return belongs inside the try, so a success leaves the loop immediately. And the final raise belongs after the loop, not inside it — otherwise you give up after the first failure.

📋 Quick Reference — REST API Clients

SyntaxWhat it does
requests.get(url, timeout=5)Make a GET request with timeout
requests.post(url, json=data)POST JSON body
response.raise_for_status()Raise exception on 4xx/5xx
requests.Session()Reuse connection pool and headers
httpx.AsyncClient()Async HTTP client

🎉 Great work! You've completed this lesson.

You can now build robust API clients with auth, retry logic, rate limiting, and async support for high-performance integrations.

Practice quiz

Which HTTP method fetches data without changing anything on the server?

  • GET
  • POST
  • DELETE
  • PUT

Answer: GET. GET retrieves resources and is safe and idempotent — it has no side effects.

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

  • GET
  • POST
  • PATCH
  • DELETE

Answer: POST. POST submits data to create new resources (like placing an order).

What is the difference between PUT and PATCH?

  • PUT replaces the resource; PATCH updates it partially
  • PUT deletes; PATCH creates
  • They are identical
  • PATCH replaces; PUT updates partially

Answer: PUT replaces the resource; PATCH updates it partially. PUT replaces the whole resource; PATCH applies a partial update.

In requests, which call makes a GET with a timeout?

  • requests.fetch(url)
  • requests.get(url, timeout=5)
  • requests.read(url)
  • requests.GET(url)

Answer: requests.get(url, timeout=5). requests.get(url, timeout=5) issues a GET and fails cleanly if the server is too slow.

How do you send a JSON body in a POST with requests?

  • requests.post(url, data=data)
  • requests.post(url, json=data)
  • requests.post(url, body=data)
  • requests.post(url, params=data)

Answer: requests.post(url, json=data). json=data serializes the dict and sets Content-Type to application/json automatically.

What does response.raise_for_status() do?

  • Prints the status code
  • Raises an exception on 4xx or 5xx responses
  • Returns True for any response
  • Retries the request

Answer: Raises an exception on 4xx or 5xx responses. It turns a failed (4xx/5xx) response into a catchable HTTPError instead of continuing silently.

A 429 status code means what?

  • Not found
  • Server error
  • Rate limited — slow down and retry
  • Success

Answer: Rate limited — slow down and retry. 429 Too Many Requests signals rate limiting; respect the Retry-After header and back off.

Which status code family indicates a SERVER error worth retrying?

  • 2xx
  • 3xx
  • 4xx
  • 5xx

Answer: 5xx. 5xx codes are server-side failures; 4xx are client errors that should not simply be retried.

What is the recommended retry strategy for transient failures?

  • Retry instantly forever
  • Exponential backoff with a max attempt limit
  • Never retry
  • Retry only POST requests

Answer: Exponential backoff with a max attempt limit. Use exponential backoff (1s, 2s, 4s...) with jitter and a cap, retrying safe operations.

Why use a requests.Session() object?

  • To make requests slower
  • For connection pooling and shared headers/cookies
  • To avoid timeouts
  • It is required for every request

Answer: For connection pooling and shared headers/cookies. A Session reuses the underlying connection pool and persists headers and cookies across calls.

Continue this course