← Python

REST APIs, HTTP & Web Scraping

← REST APIs, HTTP & Web Scraping

Select a question from the sidebar

18 questions in this topic

Question 1 / 18

What are the main HTTP methods and when should you use each one?

HTTP defines several methods (verbs) that indicate the desired action on a resource:

Method	Purpose	Idempotent	Safe	Has Body
GET	Retrieve a resource	Yes	Yes	No
POST	Create a new resource	No	No	Yes
PUT	Replace an entire resource	Yes	No	Yes
PATCH	Partially update a resource	No*	No	Yes
DELETE	Remove a resource	Yes	No	Optional
HEAD	Like GET but returns only headers	Yes	Yes	No
OPTIONS	Returns supported methods for a URL	Yes	Yes	No

Idempotent means making the same request multiple times has the same effect as making it once. GET, PUT, DELETE are idempotent. POST is not (each call may create a new resource).

Safe means the request does not modify server state. GET, HEAD, OPTIONS are safe.

*PATCH is technically not guaranteed idempotent by the spec, though implementations often make it so.

Pseudo Code

GET /users          -> List all users
GET /users/123      -> Get user 123
POST /users         -> Create a new user
PUT /users/123      -> Replace user 123 entirely
PATCH /users/123    -> Update specific fields of user 123
DELETE /users/123   -> Delete user 123
HEAD /users         -> Check if /users exists (headers only)
OPTIONS /users      -> What methods does /users support?

Code

import requests

base_url = 'https://api.example.com'

# GET - Retrieve
response = requests.get(f'{base_url}/users')
users = response.json()

response = requests.get(f'{base_url}/users/123')
user = response.json()

# POST - Create
response = requests.post(f'{base_url}/users', json={
    'name': 'Alice',
    'email': 'alice@example.com'
})
new_user = response.json()  # Includes generated id

# PUT - Full replace
response = requests.put(f'{base_url}/users/123', json={
    'name': 'Alice Updated',
    'email': 'alice_new@example.com',
    'age': 30  # Must include ALL fields
})

# PATCH - Partial update
response = requests.patch(f'{base_url}/users/123', json={
    'name': 'Alice Patched'  # Only the fields to change
})

# DELETE - Remove
response = requests.delete(f'{base_url}/users/123')

# HEAD - Headers only (no body)
response = requests.head(f'{base_url}/users')
print(response.headers['Content-Length'])

# OPTIONS - Supported methods
response = requests.options(f'{base_url}/users')
print(response.headers.get('Allow'))  # e.g., 'GET, POST, OPTIONS'

Key Points

•GET retrieves data (safe, idempotent); POST creates (not idempotent)
•PUT replaces the full resource; PATCH updates specific fields
•DELETE removes a resource; idempotent (deleting twice returns 404 the second time)
•HEAD is like GET but returns only headers (useful for checking existence or size)
•Idempotent means repeated requests have the same effect

Gotchas

•PUT requires sending the complete resource; omitting a field may set it to null
•POST is not idempotent: retrying a failed POST may create duplicate resources
•Some APIs misuse POST for everything; proper method usage improves cacheability and semantics

GET retrieves, POST creates, PUT replaces entirely, PATCH updates partially, DELETE removes. GET/PUT/DELETE are idempotent (safe to retry). POST is not idempotent. HEAD returns headers only. OPTIONS lists supported methods.

•GET/PUT/DELETE are idempotent; POST is not
•PUT replaces entire resource; PATCH updates specific fields

import requests

requests.get('https://api.example.com/users')           # List/retrieve
requests.post('https://api.example.com/users', json={})  # Create
requests.put('https://api.example.com/users/1', json={}) # Full replace
requests.patch('https://api.example.com/users/1', json={}) # Partial update
requests.delete('https://api.example.com/users/1')       # Delete

PUT omitting a field may null it; always send the complete resource

Question 2 / 18

What are HTTP status codes and which ones should you know for APIs?

HTTP status codes are three-digit numbers grouped by category:

1xx Informational

100 Continue - Server received headers, client should send body

2xx Success

200 OK - Standard success (GET, PUT, PATCH)
201 Created - Resource created successfully (POST)
204 No Content - Success with no body (DELETE)
202 Accepted - Request accepted for async processing

3xx Redirection

301 Moved Permanently - Resource URL changed permanently
302 Found - Temporary redirect
304 Not Modified - Cached version is still valid

4xx Client Errors

400 Bad Request - Malformed request or validation error
401 Unauthorized - Authentication required
403 Forbidden - Authenticated but not authorized
404 Not Found - Resource does not exist
405 Method Not Allowed - HTTP method not supported
409 Conflict - State conflict (e.g., duplicate resource)
422 Unprocessable Entity - Validation failed (well-formed but semantically wrong)
429 Too Many Requests - Rate limit exceeded

5xx Server Errors

500 Internal Server Error - Unhandled server exception
502 Bad Gateway - Upstream server error
503 Service Unavailable - Server overloaded or in maintenance
504 Gateway Timeout - Upstream server timed out

Pseudo Code

2xx = Success (200 OK, 201 Created, 204 No Content)
3xx = Redirect (301 Permanent, 302 Temporary, 304 Cached)
4xx = Client error (400 Bad Request, 401 Unauth, 403 Forbidden, 404 Not Found)
5xx = Server error (500 Internal, 502 Bad Gateway, 503 Unavailable)

Code

import requests

response = requests.get('https://api.example.com/users/123')

# Check status code
print(response.status_code)     # 200
print(response.reason)          # 'OK'
print(response.ok)              # True (status < 400)

# Raise an exception for 4xx/5xx status codes
response.raise_for_status()  # Raises HTTPError if status >= 400

# Common pattern: handle different status codes
def get_user(user_id):
    response = requests.get(f'https://api.example.com/users/{user_id}')

    if response.status_code == 200:
        return response.json()
    elif response.status_code == 404:
        return None  # User not found
    elif response.status_code == 401:
        raise PermissionError('Authentication required')
    elif response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        raise Exception(f'Rate limited. Retry after {retry_after}s')
    else:
        response.raise_for_status()  # Raise for unexpected errors

# Flask API returning appropriate status codes
from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data or 'name' not in data:
        return jsonify({'error': 'name is required'}), 400
    # ... create user ...
    return jsonify({'id': 1, 'name': data['name']}), 201

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    # ... delete user ...
    return '', 204  # No content

Key Points

•2xx = success, 3xx = redirect, 4xx = client error, 5xx = server error
•201 Created for POST success; 204 No Content for DELETE success
•401 = not authenticated; 403 = authenticated but not authorized
•response.raise_for_status() raises HTTPError for 4xx/5xx
•429 Too Many Requests often includes a Retry-After header

Gotchas

•401 means 'not authenticated' despite being named 'Unauthorized'
•Some APIs return 200 for everything and put the actual status in the body (bad practice)
•404 may mean the resource does not exist OR the URL is wrong; check the endpoint

Status codes: 2xx = success (200 OK, 201 Created, 204 No Content), 4xx = client error (400 Bad Request, 401 Unauth, 403 Forbidden, 404 Not Found, 429 Rate Limited), 5xx = server error (500, 502, 503). Use response.raise_for_status() to auto-raise on errors.

•200 OK, 201 Created, 204 No Content for success; 400/401/403/404 for client errors
•response.raise_for_status() raises on 4xx/5xx

response = requests.get('https://api.example.com/users/1')
print(response.status_code)  # 200
response.raise_for_status()  # Raises HTTPError for 4xx/5xx

# Return proper codes in Flask
return jsonify(user), 200      # OK
return jsonify(user), 201      # Created
return '', 204                 # No Content (DELETE)
return jsonify(error), 400     # Bad Request
return jsonify(error), 404     # Not Found

401 means not authenticated, not 'not authorized' (that is 403)

Question 3 / 18

How do you use the requests library to make HTTP requests in Python?

The requests library is the standard Python library for making HTTP requests. It provides a simple, elegant API for all HTTP methods.

Key features:

Simple API: requests.get(), .post(), .put(), .patch(), .delete()
Automatic JSON encoding/decoding with json= parameter
Support for headers, query parameters, form data, file uploads
Session objects for persistent settings and cookies
Timeout handling, redirect following, SSL verification

Response object attributes:

.status_code - HTTP status code
.json() - Parse response body as JSON
.text - Response body as text
.content - Response body as bytes
.headers - Response headers (case-insensitive dict)
.url - Final URL (after redirects)
.elapsed - Time taken for the request

Pseudo Code

1. import requests
2. Make a request: response = requests.get(url, params=, headers=)
3. Check response.status_code or response.ok
4. Parse body: response.json() or response.text
5. Handle errors with response.raise_for_status()

Code

import requests

# GET with query parameters
response = requests.get(
    'https://api.example.com/users',
    params={'page': 1, 'limit': 10, 'active': True},
    headers={'Accept': 'application/json'}
)
print(response.url)  # https://api.example.com/users?page=1&limit=10&active=True
users = response.json()

# POST with JSON body
response = requests.post(
    'https://api.example.com/users',
    json={'name': 'Alice', 'email': 'alice@example.com'},
    headers={'Content-Type': 'application/json'}
)
new_user = response.json()

# POST with form data
response = requests.post(
    'https://api.example.com/login',
    data={'username': 'alice', 'password': 'secret'}  # form-encoded
)

# PUT with full replacement
response = requests.put(
    'https://api.example.com/users/123',
    json={'name': 'Alice Updated', 'email': 'alice@example.com', 'age': 30}
)

# PATCH with partial update
response = requests.patch(
    'https://api.example.com/users/123',
    json={'name': 'Alice Patched'}
)

# DELETE
response = requests.delete('https://api.example.com/users/123')

# File upload
with open('photo.jpg', 'rb') as f:
    response = requests.post(
        'https://api.example.com/upload',
        files={'image': ('photo.jpg', f, 'image/jpeg')}
    )

# Response inspection
print(response.status_code)          # 200
print(response.headers['Content-Type'])  # application/json
print(response.elapsed.total_seconds())  # 0.342
print(response.text)                 # Raw text body
print(response.content)              # Raw bytes body
print(response.encoding)             # utf-8

# Error handling
try:
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    print(f'HTTP error: {e}')
except requests.exceptions.ConnectionError:
    print('Connection failed')
except requests.exceptions.Timeout:
    print('Request timed out')

Key Points

•json= auto-serializes dict to JSON and sets Content-Type; data= sends form-encoded
•params= adds query string parameters to the URL
•.json() parses response body; .text for raw string; .content for bytes
•response.raise_for_status() raises HTTPError for 4xx/5xx
•files= for multipart file uploads

Gotchas

•json= and data= are mutually exclusive; using both causes unexpected behavior
•response.json() raises ValueError if the body is not valid JSON; wrap in try/except
•requests does not set a default timeout; always pass timeout= to avoid hanging

Use requests.get(url, params=) for GET, requests.post(url, json=) for POST with JSON body, data= for form data. Access response with .json(), .text, .status_code. Always use timeout= and raise_for_status().

•json= for JSON body; data= for form data; params= for query string
•Always set timeout= and use raise_for_status()

import requests

# GET with params
resp = requests.get('https://api.example.com/users', params={'page': 1}, timeout=10)
users = resp.json()

# POST with JSON
resp = requests.post('https://api.example.com/users',
                     json={'name': 'Alice'}, timeout=10)
resp.raise_for_status()
new_user = resp.json()

Without timeout=, requests can hang indefinitely

Question 4 / 18

How do you handle headers and authentication in HTTP requests?

Headers are key-value metadata sent with HTTP requests and responses. Common request headers include Authorization, Content-Type, Accept, User-Agent.

Authentication methods:

Basic Auth: Base64-encoded username:password in the Authorization header. Use requests' built-in support.
Bearer Token: A token (JWT, OAuth, API key) in the Authorization header.
API Key: Sent as a header (e.g., X-API-Key) or as a query parameter.
OAuth 2.0: Token-based auth flow. Typically involves getting a token from an auth server first.

The requests library supports all of these through the headers= parameter, auth= parameter, or Session objects.

Pseudo Code

1. Set custom headers with headers={'Header-Name': 'value'}
2. For Basic Auth: use auth=('username', 'password')
3. For Bearer tokens: set Authorization header
4. For OAuth: get token first, then send as Bearer
5. Use Session() to persist auth across multiple requests

Code

import requests
from requests.auth import HTTPBasicAuth

# Custom headers
response = requests.get(
    'https://api.example.com/data',
    headers={
        'Accept': 'application/json',
        'User-Agent': 'MyApp/1.0',
        'X-Custom-Header': 'some-value',
    }
)

# Basic Authentication
response = requests.get(
    'https://api.example.com/protected',
    auth=('username', 'password')  # Shorthand for HTTPBasicAuth
)
# Or explicitly:
response = requests.get(
    'https://api.example.com/protected',
    auth=HTTPBasicAuth('username', 'password')
)

# Bearer Token (JWT, OAuth)
token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'
response = requests.get(
    'https://api.example.com/me',
    headers={'Authorization': f'Bearer {token}'}
)

# API Key in header
response = requests.get(
    'https://api.example.com/data',
    headers={'X-API-Key': 'your-api-key-here'}
)

# API Key in query parameter
response = requests.get(
    'https://api.example.com/data',
    params={'api_key': 'your-api-key-here'}
)

# OAuth 2.0 token flow
def get_oauth_token(client_id, client_secret):
    response = requests.post(
        'https://auth.example.com/oauth/token',
        data={
            'grant_type': 'client_credentials',
            'client_id': client_id,
            'client_secret': client_secret,
        }
    )
    response.raise_for_status()
    return response.json()['access_token']

token = get_oauth_token('my_client_id', 'my_client_secret')
response = requests.get(
    'https://api.example.com/resource',
    headers={'Authorization': f'Bearer {token}'}
)

# Persistent auth with Session
session = requests.Session()
session.headers.update({
    'Authorization': f'Bearer {token}',
    'Accept': 'application/json',
})

# All requests through this session use the same headers
users = session.get('https://api.example.com/users').json()
orders = session.get('https://api.example.com/orders').json()
session.close()

Key Points

•auth=('user', 'pass') for Basic Auth; headers={'Authorization': 'Bearer token'} for tokens
•Session objects persist headers, cookies, and auth across multiple requests
•API keys can go in headers (X-API-Key) or query params (params={'api_key': ...})
•OAuth 2.0 typically requires a token exchange step before making API calls
•Response headers are case-insensitive: response.headers['content-type'] works

Gotchas

•Never hardcode credentials in source code; use environment variables or secret managers
•Basic Auth sends credentials base64-encoded (not encrypted); always use HTTPS
•Bearer tokens expire; implement token refresh logic for long-running applications

Set headers with headers={}. Use auth=('user', 'pass') for Basic Auth, headers={'Authorization': 'Bearer token'} for tokens. Use Session() to persist auth across requests.

•auth= for Basic; Authorization header for Bearer tokens
•Session persists headers and cookies across calls

import requests

# Basic Auth
resp = requests.get(url, auth=('user', 'pass'))

# Bearer Token
resp = requests.get(url, headers={'Authorization': f'Bearer {token}'})

# Session for persistent auth
session = requests.Session()
session.headers['Authorization'] = f'Bearer {token}'
resp = session.get(url)

Never hardcode secrets; use environment variables

Question 5 / 18

How do you handle JSON data in HTTP requests and responses?

JSON (JavaScript Object Notation) is the standard data format for REST APIs. Python's json module and the requests library provide seamless JSON support.

Sending JSON:

requests.post(url, json=data) - Automatically serializes data to JSON and sets Content-Type: application/json.
requests.post(url, data=json.dumps(data)) - Manual serialization (you must set Content-Type yourself).

Receiving JSON:

response.json() - Parses the response body as JSON into a Python dict/list.

JSON mapping:

JSON	Python
object {}	dict
array []	list
string	str
number (int)	int
number (float)	float
true/false	True/False
null	None

Pseudo Code

1. Send JSON: requests.post(url, json=dict) (auto-serializes)
2. Receive JSON: response.json() (auto-deserializes)
3. For custom serialization, use json.dumps() / json.loads()
4. Handle non-JSON responses with try/except on .json()
5. Use json.dumps(data, indent=2) for pretty printing

Code

import json
import requests
from datetime import datetime

# Sending JSON with requests
response = requests.post(
    'https://api.example.com/users',
    json={'name': 'Alice', 'age': 30, 'active': True}  # Auto-serialized
)

# Receiving JSON
data = response.json()  # Parse JSON body into dict
print(data['name'])      # Alice

# Safe JSON parsing
try:
    data = response.json()
except requests.exceptions.JSONDecodeError:
    print('Response is not valid JSON')
    print(response.text)  # Inspect the raw response

# Manual JSON handling with json module
user_dict = {'name': 'Alice', 'age': 30}
json_string = json.dumps(user_dict)          # dict -> JSON string
parsed_dict = json.loads(json_string)        # JSON string -> dict

# Pretty printing
print(json.dumps(user_dict, indent=2))
# {
#   "name": "Alice",
#   "age": 30
# }

# Custom serialization for non-standard types
def json_serializer(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, set):
        return list(obj)
    raise TypeError(f'Type {type(obj)} is not JSON serializable')

data = {
    'name': 'Alice',
    'created': datetime.now(),
    'tags': {'python', 'developer'},
}
json_string = json.dumps(data, default=json_serializer)
print(json_string)
# {"name": "Alice", "created": "2024-01-15T10:30:00", "tags": ["python", "developer"]}

# Reading/writing JSON files
with open('data.json', 'w') as f:
    json.dump(user_dict, f, indent=2)

with open('data.json', 'r') as f:
    loaded = json.load(f)

Key Points

•json= parameter auto-serializes and sets Content-Type; data= does not
•response.json() parses the body; wrap in try/except for safety
•json.dumps() converts Python to JSON string; json.loads() reverses it
•json.dump()/json.load() work with file objects directly
•Use default= parameter in json.dumps() for custom type serialization

Gotchas

•datetime, set, and bytes are not JSON-serializable by default; use a custom serializer
•json.dumps() outputs strings, not bytes; use .encode() if bytes are needed
•JSON keys must be strings; dict keys that are int will be converted to strings

Use json=dict in requests to auto-serialize and set Content-Type. Use response.json() to parse responses. For manual work, json.dumps() serializes and json.loads() parses. Custom types need a default= function.

•json= auto-serializes; response.json() auto-parses
•json.dumps/loads for manual string conversion

import requests, json

# Send JSON
resp = requests.post(url, json={'name': 'Alice'})

# Receive JSON
data = resp.json()

# Manual
json_str = json.dumps({'name': 'Alice'}, indent=2)
parsed = json.loads(json_str)

datetime and set are not JSON-serializable without a custom default function

Question 6 / 18

How do you use request Sessions for connection reuse and persistent settings?

A requests.Session() object provides:

Connection pooling - Reuses TCP connections to the same host, significantly faster for multiple requests.
Persistent settings - Headers, auth, cookies, and other parameters persist across requests.
Cookie handling - Cookies received in responses are automatically sent with subsequent requests.

Sessions are especially useful when making many requests to the same API, as they avoid the overhead of creating a new TCP connection for each request.

Pseudo Code

1. Create session = requests.Session()
2. Set persistent headers with session.headers.update()
3. Set auth with session.auth = ('user', 'pass')
4. Make requests: session.get(), session.post()
5. Cookies are automatically persisted
6. Close with session.close() or use as context manager

Code

import requests

# Without Session: each request creates a new TCP connection
for i in range(100):
    requests.get(f'https://api.example.com/items/{i}')  # Slow!

# With Session: connections are reused
with requests.Session() as session:
    # Persistent settings
    session.headers.update({
        'Authorization': 'Bearer my-token',
        'Accept': 'application/json',
        'User-Agent': 'MyApp/1.0',
    })
    session.auth = ('user', 'pass')  # For Basic Auth

    # All requests reuse the TCP connection and include these headers
    for i in range(100):
        resp = session.get(f'https://api.example.com/items/{i}')  # Fast!

    # Cookies persist automatically
    session.post('https://example.com/login', data={'user': 'alice', 'pass': 'secret'})
    # Subsequent requests include the login cookie
    profile = session.get('https://example.com/profile').json()

    # Per-request headers merge with session headers
    resp = session.get(
        'https://api.example.com/special',
        headers={'X-Special': 'value'}  # Added to session headers
    )

    # Inspect cookies
    print(session.cookies.get_dict())

# Building a reusable API client
class APIClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers['X-API-Key'] = api_key
        self.session.headers['Accept'] = 'application/json'

    def get_users(self):
        resp = self.session.get(f'{self.base_url}/users', timeout=10)
        resp.raise_for_status()
        return resp.json()

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

    def __enter__(self):
        return self

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

with APIClient('https://api.example.com', 'my-key') as client:
    users = client.get_users()

Key Points

•Session reuses TCP connections (connection pooling) for massive speed improvement
•Headers, auth, and cookies persist across all requests in a session
•Per-request headers merge with (not replace) session headers
•Use as a context manager (with Session() as s:) for automatic cleanup
•Ideal for building reusable API client classes

Gotchas

•Forgetting to close a Session can leak connections; use context managers
•Session is not thread-safe; use one session per thread
•Per-request headers override session headers with the same key, not merge them

requests.Session() reuses TCP connections and persists headers, auth, and cookies. Much faster for multiple requests to the same host. Use as a context manager for auto-cleanup.

•Connection pooling + persistent headers/cookies
•Use context manager for cleanup; not thread-safe

with requests.Session() as session:
    session.headers['Authorization'] = 'Bearer token'
    resp1 = session.get('https://api.example.com/users')   # Reuses connection
    resp2 = session.get('https://api.example.com/orders')  # Same connection

Session is not thread-safe; use one per thread

Question 7 / 18

How do you handle timeouts and retries in HTTP requests?

Timeouts prevent your program from hanging indefinitely when a server is slow or unresponsive. requests supports two timeout values:

connect timeout: How long to wait for the TCP connection to be established.
read timeout: How long to wait for the server to send data.

Retries handle transient failures (network blips, server overload). The requests library does not retry by default, but you can configure automatic retries using urllib3.util.retry.Retry with a custom adapter.

Common retry strategies:

Retry on specific status codes (429, 500, 502, 503, 504)
Exponential backoff between retries
Maximum retry count

Pseudo Code

1. Always set timeout=(connect, read) on every request
2. Create a Retry object with total retries, backoff, status forcelist
3. Mount the retry adapter on a Session
4. All requests through the session will auto-retry on failure

Code

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Basic timeout
response = requests.get(
    'https://api.example.com/data',
    timeout=5  # 5 seconds for both connect and read
)

# Separate connect and read timeouts
response = requests.get(
    'https://api.example.com/data',
    timeout=(3.05, 27)  # (connect_timeout, read_timeout)
)

# Handling timeout exceptions
try:
    response = requests.get('https://api.example.com/slow', timeout=5)
except requests.exceptions.ConnectTimeout:
    print('Connection timed out')
except requests.exceptions.ReadTimeout:
    print('Server took too long to respond')
except requests.exceptions.Timeout:
    print('Request timed out (generic)')

# Automatic retries with exponential backoff
retry_strategy = Retry(
    total=3,                   # Max 3 retries
    backoff_factor=1,          # Wait 1s, 2s, 4s between retries
    status_forcelist=[429, 500, 502, 503, 504],  # Retry on these codes
    allowed_methods=['GET', 'POST', 'PUT'],       # Which methods to retry
    raise_on_status=False,     # Don't raise on exhausted retries
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount('https://', adapter)
session.mount('http://', adapter)

# All requests now auto-retry on failure
response = session.get('https://api.example.com/data', timeout=10)

# Manual retry with custom logic
import time

def request_with_retry(url, max_retries=3, backoff=1):
    for attempt in range(max_retries + 1):
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 429:
                wait = int(response.headers.get('Retry-After', backoff * (2 ** attempt)))
                time.sleep(wait)
                continue
            response.raise_for_status()
            return response
        except requests.exceptions.ConnectionError:
            if attempt == max_retries:
                raise
            time.sleep(backoff * (2 ** attempt))
    raise requests.exceptions.RetryError('Max retries exceeded')

Key Points

•Always set timeout=; without it, requests can hang forever
•timeout=(connect, read) allows separate connect and read timeouts
•Retry with backoff_factor for exponential wait (1s, 2s, 4s...)
•status_forcelist specifies which HTTP codes trigger a retry
•Mount adapters on both http:// and https:// for complete coverage

Gotchas

•Retry on POST can cause duplicate resources if the server processes the request but the response is lost
•backoff_factor=1 means waits of 0s, 1s, 2s, 4s (first retry is immediate)
•urllib3 Retry does not retry on all exceptions by default; only ConnectionError and some others

Always set timeout= (default is None, which hangs forever). Use urllib3.util.retry.Retry with HTTPAdapter for automatic retries with exponential backoff on specific status codes.

•timeout=(connect, read) prevents hanging; Retry auto-retries on failure
•backoff_factor=1 gives exponential wait: 1s, 2s, 4s...

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503])
session = requests.Session()
session.mount('https://', HTTPAdapter(max_retries=retry))

resp = session.get('https://api.example.com/data', timeout=(3, 10))

Retrying POST can create duplicates if the server already processed the request

Question 8 / 18

How do you handle query parameters and pagination in REST API requests?

Query parameters are key-value pairs appended to the URL after ?. They are used for filtering, sorting, searching, and pagination.

Common pagination patterns:

Offset/Limit: ?page=2&limit=20 or ?offset=20&limit=20
Cursor-based: ?cursor=eyJpZCI6MTAwfQ== (encoded position marker)
Link header: The API returns Link: <url>; rel="next" headers

Cursor-based pagination is preferred for large datasets because offset pagination degrades as offset grows (the DB must skip rows).

Pseudo Code

1. Pass query params with params= dict (auto-encoded)
2. For offset pagination: increment page number until no more results
3. For cursor pagination: follow the cursor/next_url from the response
4. For Link headers: parse the Link header for the next URL

Code

import requests

# Sending query parameters
response = requests.get(
    'https://api.example.com/users',
    params={
        'search': 'alice',
        'role': 'admin',
        'sort': '-created_at',  # Descending by convention
        'page': 1,
        'per_page': 25,
    }
)
# URL: https://api.example.com/users?search=alice&role=admin&sort=-created_at&page=1&per_page=25

# Offset-based pagination (collect all pages)
def fetch_all_offset(base_url, per_page=100):
    all_items = []
    page = 1
    while True:
        response = requests.get(
            base_url,
            params={'page': page, 'per_page': per_page},
            timeout=10
        )
        response.raise_for_status()
        data = response.json()

        items = data.get('results', [])
        if not items:
            break

        all_items.extend(items)
        page += 1

        # Alternative: check total count
        if len(all_items) >= data.get('total', float('inf')):
            break

    return all_items

all_users = fetch_all_offset('https://api.example.com/users')

# Cursor-based pagination
def fetch_all_cursor(base_url, per_page=100):
    all_items = []
    cursor = None
    while True:
        params = {'per_page': per_page}
        if cursor:
            params['cursor'] = cursor

        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()

        all_items.extend(data['results'])

        cursor = data.get('next_cursor')
        if not cursor:
            break

    return all_items

# Link header pagination (GitHub-style)
def fetch_all_link_header(url):
    all_items = []
    while url:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        all_items.extend(response.json())

        # Parse Link header
        url = None
        link_header = response.headers.get('Link', '')
        for part in link_header.split(','):
            if 'rel="next"' in part:
                url = part.split(';')[0].strip(' <>')

    return all_items

repos = fetch_all_link_header('https://api.github.com/users/octocat/repos')

Key Points

•params= dict auto-encodes and appends query parameters to the URL
•Offset pagination: simple but slow for large datasets (DB skips rows)
•Cursor pagination: scalable, uses an opaque marker for position
•Link headers provide hypermedia-driven pagination (no need to construct URLs)
•Always check for empty results or missing next cursor to stop pagination

Gotchas

•Offset pagination can return duplicates or skip items if data changes between pages
•Some APIs use 'next' URL in the JSON body instead of Link headers
•params= with None values are sent as the string 'None'; filter them out before passing

Use params=dict to add query parameters. For pagination, loop through pages (offset-based) or follow cursors/next URLs. Cursor-based pagination is preferred for large datasets.

•params= auto-encodes query string; loop until empty for pagination
•Cursor pagination is more reliable than offset for large/changing datasets

# Query params
resp = requests.get(url, params={'page': 1, 'per_page': 25, 'sort': 'name'})

# Paginate all pages
all_items = []
page = 1
while True:
    data = requests.get(url, params={'page': page}).json()
    if not data['results']:
        break
    all_items.extend(data['results'])
    page += 1

Offset pagination can skip/duplicate items if data changes between page fetches

Question 9 / 18

What are the key REST API design principles?

REST (Representational State Transfer) is an architectural style for APIs. Key principles:

Resource-based URLs: URLs represent nouns (resources), not actions.
- Good: GET /users/123 | Bad: GET /getUser?id=123
HTTP methods as verbs: GET=read, POST=create, PUT=replace, PATCH=update, DELETE=remove.
Stateless: Each request contains all the information needed. No server-side session state.
Consistent naming:
- Plural nouns: /users, /orders
- Nested resources: /users/123/orders
- Lowercase, hyphen-separated: /user-profiles
Standard status codes: Use proper HTTP status codes (201 for created, 404 for not found, etc.).
Filtering, sorting, pagination: Use query parameters.
HATEOAS (Hypermedia): Responses include links to related resources.
Versioning: Version your API to avoid breaking changes.

Pseudo Code

Resource design:
GET    /users          -> List users
GET    /users/123      -> Get user 123
POST   /users          -> Create user
PUT    /users/123      -> Replace user 123
PATCH  /users/123      -> Update user 123
DELETE /users/123      -> Delete user 123
GET    /users/123/orders -> List orders for user 123

Code

from flask import Flask, jsonify, request, url_for

app = Flask(__name__)

# Good REST API design in Flask
@app.route('/api/v1/users', methods=['GET'])
def list_users():
    """List users with filtering, sorting, and pagination."""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    sort = request.args.get('sort', 'name')
    role = request.args.get('role')  # Filter

    # ... fetch from database ...
    users = [{'id': 1, 'name': 'Alice', 'role': 'admin'}]
    total = 100

    return jsonify({
        'data': users,
        'meta': {
            'page': page,
            'per_page': per_page,
            'total': total,
            'total_pages': (total + per_page - 1) // per_page,
        },
        'links': {
            'self': f'/api/v1/users?page={page}',
            'next': f'/api/v1/users?page={page+1}' if page * per_page < total else None,
            'prev': f'/api/v1/users?page={page-1}' if page > 1 else None,
        }
    })

@app.route('/api/v1/users', methods=['POST'])
def create_user():
    """Create a new user."""
    data = request.get_json()
    if not data or 'name' not in data:
        return jsonify({'error': 'name is required'}), 400

    user = {'id': 2, 'name': data['name']}
    return jsonify({'data': user}), 201

@app.route('/api/v1/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """Get a specific user."""
    user = None  # ... fetch from DB ...
    if not user:
        return jsonify({'error': 'User not found'}), 404
    return jsonify({'data': user})

@app.route('/api/v1/users/<int:user_id>', methods=['PATCH'])
def update_user(user_id):
    """Partially update a user."""
    data = request.get_json()
    # ... update only provided fields ...
    return jsonify({'data': {'id': user_id, **data}})

@app.route('/api/v1/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    """Delete a user."""
    # ... delete from DB ...
    return '', 204

# Nested resources
@app.route('/api/v1/users/<int:user_id>/orders', methods=['GET'])
def list_user_orders(user_id):
    """List orders for a specific user."""
    return jsonify({'data': [], 'meta': {'user_id': user_id}})

# Consistent error format
@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Resource not found', 'status': 404}), 404

@app.errorhandler(500)
def internal_error(error):
    return jsonify({'error': 'Internal server error', 'status': 500}), 500

Key Points

•URLs are nouns (resources), HTTP methods are verbs (actions)
•Use plural nouns: /users not /user; nested: /users/123/orders
•Consistent response format: wrap data in 'data' key with 'meta' for pagination
•Proper status codes: 201 Created, 204 No Content, 400 Bad Request, 404 Not Found
•Version your API in the URL (/api/v1/) or via Accept header

Gotchas

•Avoid verbs in URLs: /users not /getUsers or /createUser
•PATCH is for partial updates; PUT requires sending the complete resource
•Overly nested URLs (/users/1/orders/2/items/3) become hard to maintain; limit to 2 levels

REST uses resource-based URLs (nouns) with HTTP methods (verbs). Use plural nouns (/users), proper status codes, query params for filtering/sorting/pagination, and consistent response format. Version your API (/api/v1/).

•Nouns for URLs, HTTP methods for actions, plural resources
•Proper status codes and consistent error format

# REST resource pattern
# GET    /api/v1/users          -> List
# POST   /api/v1/users          -> Create (returns 201)
# GET    /api/v1/users/123      -> Read
# PATCH  /api/v1/users/123      -> Update
# DELETE /api/v1/users/123      -> Delete (returns 204)
# GET    /api/v1/users/123/orders -> Nested resource

Avoid verbs in URLs and limit nesting to 2 levels

Question 10 / 18

What is CORS and how do you handle it in Python web applications?

CORS (Cross-Origin Resource Sharing) is a browser security mechanism that restricts web pages from making requests to a different domain than the one that served the page.

When a frontend at http://frontend.com tries to call http://api.backend.com, the browser sends a preflight OPTIONS request to check if the API allows cross-origin requests. The API must respond with specific CORS headers.

Key CORS headers:

Access-Control-Allow-Origin: Which origins can access the resource (* for all, or specific domain)
Access-Control-Allow-Methods: Allowed HTTP methods (GET, POST, etc.)
Access-Control-Allow-Headers: Allowed request headers (Authorization, Content-Type, etc.)
Access-Control-Allow-Credentials: Whether cookies/auth are allowed (true/false)
Access-Control-Max-Age: How long the preflight result can be cached

When does a preflight occur?

Methods other than GET, HEAD, POST
Custom headers (Authorization, X-API-Key)
Content-Type other than form-encoded, multipart, or text/plain

Pseudo Code

1. Browser sends OPTIONS preflight to the API
2. API responds with CORS headers
3. Browser checks if the origin is allowed
4. If allowed, browser sends the actual request
5. If not allowed, browser blocks the request (CORS error)

Code

# Method 1: Flask-CORS extension (recommended)
from flask import Flask
from flask_cors import CORS

app = Flask(__name__)

# Allow all origins (development only!)
CORS(app)

# Allow specific origins (production)
CORS(app, resources={
    r'/api/*': {
        'origins': ['https://myapp.com', 'https://staging.myapp.com'],
        'methods': ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'],
        'allow_headers': ['Authorization', 'Content-Type'],
        'supports_credentials': True,
        'max_age': 600,
    }
})

# Method 2: Manual CORS headers
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.after_request
def add_cors_headers(response):
    origin = request.headers.get('Origin')
    allowed_origins = ['https://myapp.com', 'https://staging.myapp.com']

    if origin in allowed_origins:
        response.headers['Access-Control-Allow-Origin'] = origin
        response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, PATCH, DELETE, OPTIONS'
        response.headers['Access-Control-Allow-Headers'] = 'Authorization, Content-Type'
        response.headers['Access-Control-Allow-Credentials'] = 'true'
        response.headers['Access-Control-Max-Age'] = '600'

    return response

# Handle preflight OPTIONS request explicitly
@app.route('/api/data', methods=['OPTIONS'])
def preflight():
    return '', 204

# Method 3: Per-route CORS with decorator
from flask_cors import cross_origin

@app.route('/api/public', methods=['GET'])
@cross_origin(origins='*')  # Public endpoint
def public_data():
    return jsonify({'data': 'public'})

@app.route('/api/private', methods=['GET'])
@cross_origin(origins=['https://myapp.com'], supports_credentials=True)
def private_data():
    return jsonify({'data': 'private'})

Key Points

•CORS is a browser security feature; it does not affect server-to-server requests
•Preflight OPTIONS requests check if cross-origin requests are allowed
•Flask-CORS is the easiest way to handle CORS in Flask
•Access-Control-Allow-Origin: * cannot be used with credentials (cookies/auth)
•max_age caches the preflight result so the browser does not re-check every request

Gotchas

•CORS errors only appear in the browser; curl and Postman work fine without CORS
•Using '*' for Allow-Origin with credentials=True is not allowed by the spec
•Forgetting to handle OPTIONS requests causes preflight failures
•CORS headers must be set on the response, not the request

CORS is a browser security mechanism blocking cross-origin requests. Fix by adding Access-Control-Allow-Origin headers on the server. In Flask, use flask-cors: CORS(app, origins=['https://myapp.com']). Remember CORS only affects browsers, not server-to-server calls.

•CORS is browser-only; use flask-cors for easy setup
•Cannot use '*' origin with credentials

from flask import Flask
from flask_cors import CORS

app = Flask(__name__)
CORS(app, resources={r'/api/*': {'origins': ['https://myapp.com']}})

CORS errors only show in browsers; curl/Postman ignore CORS

Question 11 / 18

What are webhooks and how do you implement them in Python?

A webhook is an HTTP callback: when an event occurs in one system, it sends an HTTP POST request to a URL configured by another system. Instead of polling an API repeatedly, you receive push notifications.

How it works:

You provide a callback URL to the service (e.g., https://myapp.com/webhooks/stripe).
When an event happens (payment completed, PR merged, etc.), the service POSTs a JSON payload to your URL.
Your server processes the payload and responds with 200 OK.

Key considerations:

Verification: Validate the webhook signature to ensure it is from the expected sender.
Idempotency: Webhooks may be sent multiple times; handle duplicates.
Quick response: Return 200 immediately and process asynchronously if needed.
Retry logic: Services retry if they do not get a 200 response.

Pseudo Code

1. Create an endpoint to receive webhook POST requests
2. Verify the webhook signature (HMAC or similar)
3. Parse the JSON payload
4. Return 200 immediately
5. Process the event asynchronously if it takes time
6. Handle duplicate deliveries with idempotency keys

Code

import hmac
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)

WEBHOOK_SECRET = 'your-webhook-secret'

def verify_signature(payload, signature, secret):
    """Verify webhook signature using HMAC-SHA256."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route('/webhooks/payment', methods=['POST'])
def handle_payment_webhook():
    # Step 1: Verify signature
    signature = request.headers.get('X-Webhook-Signature', '')
    if not verify_signature(request.data, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401

    # Step 2: Parse payload
    event = request.get_json()
    event_type = event.get('type')
    event_id = event.get('id')

    # Step 3: Idempotency check
    # if already_processed(event_id):
    #     return jsonify({'status': 'already processed'}), 200

    # Step 4: Handle the event
    if event_type == 'payment.completed':
        amount = event['data']['amount']
        user_id = event['data']['user_id']
        print(f'Payment of {amount} received for user {user_id}')
        # process_payment(event['data'])
    elif event_type == 'payment.failed':
        print(f'Payment failed: {event["data"]}')
    else:
        print(f'Unhandled event type: {event_type}')

    # Step 5: Return 200 quickly
    return jsonify({'status': 'received'}), 200

# Sending webhooks (from the other side)
import requests
import json

def send_webhook(url, payload, secret):
    """Send a webhook with HMAC signature."""
    body = json.dumps(payload)
    signature = hmac.new(
        secret.encode(),
        body.encode(),
        hashlib.sha256
    ).hexdigest()

    response = requests.post(
        url,
        json=payload,
        headers={
            'X-Webhook-Signature': signature,
            'Content-Type': 'application/json',
        },
        timeout=10
    )
    return response.status_code == 200

send_webhook(
    'https://myapp.com/webhooks/payment',
    {'type': 'payment.completed', 'id': 'evt_123', 'data': {'amount': 99.99, 'user_id': 1}},
    WEBHOOK_SECRET
)

Key Points

•Webhooks are push-based HTTP callbacks; the server sends POST requests when events occur
•Always verify webhook signatures (HMAC) to prevent spoofing
•Return 200 immediately; process asynchronously for long operations
•Implement idempotency to handle duplicate deliveries
•Use hmac.compare_digest() for timing-safe comparison

Gotchas

•Webhook endpoints must be publicly accessible (not localhost in production)
•If you do not return 200 quickly, the sender may time out and retry
•request.data must be used for signature verification (raw bytes), not request.get_json()
•Testing webhooks locally requires a tunnel tool like ngrok

Webhooks are HTTP POST callbacks triggered by events. Create an endpoint, verify the HMAC signature, process the payload, and return 200 immediately. Handle duplicate deliveries with idempotency checks.

•Verify HMAC signature, return 200 quickly, handle duplicates
•Push-based alternative to polling

import hmac, hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhooks/event', methods=['POST'])
def webhook():
    sig = request.headers.get('X-Webhook-Signature', '')
    expected = hmac.new(b'secret', request.data, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig):
        return '', 401
    event = request.get_json()
    # Process event...
    return '', 200

Use request.data (raw bytes) for signature verification, not request.get_json()

Question 12 / 18

How do you version a REST API and why is it important?

API versioning lets you evolve your API without breaking existing clients. When you make breaking changes (removing fields, changing response format, renaming endpoints), clients using the old version continue to work.

Versioning strategies:

URL path versioning (most common): /api/v1/users, /api/v2/users
Query parameter: /api/users?version=2
Header versioning: Accept: application/vnd.myapp.v2+json
Content negotiation: Custom media types in Accept header

URL path versioning is the most popular because it is explicit, easy to understand, and easy to route.

Pseudo Code

1. Start with /api/v1/ for all endpoints
2. When breaking changes are needed, create /api/v2/
3. Keep v1 running for a deprecation period
4. Use Blueprints in Flask to organize versions
5. Document the deprecation timeline

Code

from flask import Flask, Blueprint, jsonify

app = Flask(__name__)

# Version 1 Blueprint
api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')

@api_v1.route('/users', methods=['GET'])
def v1_list_users():
    """V1: Returns simple user list."""
    return jsonify({
        'users': [
            {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}
        ]
    })

# Version 2 Blueprint (breaking change: different response format)
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

@api_v2.route('/users', methods=['GET'])
def v2_list_users():
    """V2: Returns paginated user list with metadata."""
    return jsonify({
        'data': [
            {'id': 1, 'full_name': 'Alice Smith', 'email': 'alice@example.com',
             'role': 'admin'}
        ],
        'meta': {'page': 1, 'total': 1},
        'links': {'next': None}
    })

app.register_blueprint(api_v1)
app.register_blueprint(api_v2)

# Header-based versioning (alternative)
@app.route('/api/users', methods=['GET'])
def list_users_header_versioned():
    version = request.headers.get('API-Version', '1')
    if version == '2':
        return v2_list_users()
    return v1_list_users()

# Deprecation warning header
@api_v1.after_request
def add_deprecation_header(response):
    response.headers['Deprecation'] = 'true'
    response.headers['Sunset'] = 'Sat, 01 Jan 2025 00:00:00 GMT'
    response.headers['Link'] = '</api/v2>; rel="successor-version"'
    return response

# Shared logic between versions
def get_users_from_db():
    """Shared database logic."""
    return [{'id': 1, 'first_name': 'Alice', 'last_name': 'Smith', 'email': 'alice@example.com'}]

# V1 transforms the data one way, V2 another way
def format_user_v1(user):
    return {'id': user['id'], 'name': user['first_name'], 'email': user['email']}

def format_user_v2(user):
    return {'id': user['id'], 'full_name': f"{user['first_name']} {user['last_name']}",
            'email': user['email']}

Key Points

•URL path versioning (/api/v1/) is the most common and explicit approach
•Use Flask Blueprints to organize versions cleanly
•Share business logic between versions; only the presentation layer changes
•Add Deprecation and Sunset headers to old versions
•Breaking changes = different version; non-breaking changes = same version

Gotchas

•Adding new fields to a response is NOT a breaking change; removing/renaming is
•Running many API versions in parallel increases maintenance burden
•Forgetting to deprecate old versions leads to indefinite support obligations

Version your API to avoid breaking existing clients. Most common: URL path versioning (/api/v1/, /api/v2/). Use Flask Blueprints for clean organization. Share business logic, only change presentation between versions.

•URL path versioning with Blueprints is cleanest
•Adding fields is not breaking; removing/renaming is

from flask import Flask, Blueprint

api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

@api_v1.route('/users')
def v1_users():
    return jsonify({'users': [...]})

@api_v2.route('/users')
def v2_users():
    return jsonify({'data': [...], 'meta': {...}})

app.register_blueprint(api_v1)
app.register_blueprint(api_v2)

Maintain as few versions as possible; deprecate old ones with Sunset headers

Question 13 / 18

How do you handle errors properly in REST APIs?

Good API error handling provides clients with enough information to understand and fix the problem. Best practices:

Use proper HTTP status codes (400, 401, 403, 404, 422, 500).
Consistent error response format with error code, message, and details.
Never expose internal details (stack traces, SQL queries) in production.
Validate input and return specific field-level errors.
Catch all exceptions with a global error handler to prevent 500s.

Pseudo Code

1. Define a standard error response format
2. Create custom exception classes
3. Register Flask error handlers for each exception type
4. Validate input and raise specific exceptions
5. Log errors server-side, return safe messages client-side

Code

from flask import Flask, jsonify, request
from werkzeug.exceptions import HTTPException

app = Flask(__name__)

# Custom exception classes
class APIError(Exception):
    def __init__(self, message, status_code=400, error_code=None, details=None):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code or 'BAD_REQUEST'
        self.details = details

class NotFoundError(APIError):
    def __init__(self, resource, resource_id):
        super().__init__(
            message=f'{resource} with id {resource_id} not found',
            status_code=404,
            error_code='NOT_FOUND'
        )

class ValidationError(APIError):
    def __init__(self, errors):
        super().__init__(
            message='Validation failed',
            status_code=422,
            error_code='VALIDATION_ERROR',
            details=errors
        )

# Global error handlers
@app.errorhandler(APIError)
def handle_api_error(error):
    response = {
        'error': {
            'code': error.error_code,
            'message': error.message,
        }
    }
    if error.details:
        response['error']['details'] = error.details
    return jsonify(response), error.status_code

@app.errorhandler(HTTPException)
def handle_http_error(error):
    return jsonify({
        'error': {
            'code': error.name.upper().replace(' ', '_'),
            'message': error.description,
        }
    }), error.code

@app.errorhandler(Exception)
def handle_unexpected_error(error):
    # Log the actual error server-side
    app.logger.exception('Unexpected error')
    # Return generic message to client
    return jsonify({
        'error': {
            'code': 'INTERNAL_ERROR',
            'message': 'An unexpected error occurred',
        }
    }), 500

# Input validation
def validate_user_input(data):
    errors = []
    if not data:
        raise ValidationError([{'field': 'body', 'message': 'Request body is required'}])
    if not data.get('name'):
        errors.append({'field': 'name', 'message': 'Name is required'})
    if not data.get('email'):
        errors.append({'field': 'email', 'message': 'Email is required'})
    elif '@' not in data['email']:
        errors.append({'field': 'email', 'message': 'Invalid email format'})
    if errors:
        raise ValidationError(errors)

@app.route('/api/v1/users', methods=['POST'])
def create_user():
    data = request.get_json()
    validate_user_input(data)  # Raises ValidationError if invalid
    # ... create user ...
    return jsonify({'data': {'id': 1, 'name': data['name']}}), 201

@app.route('/api/v1/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = None  # ... fetch from DB ...
    if not user:
        raise NotFoundError('User', user_id)
    return jsonify({'data': user})

Key Points

•Consistent error format: {error: {code, message, details}}
•Custom exception classes map to specific HTTP status codes
•Global error handlers catch all exceptions and return proper JSON
•Field-level validation errors help clients fix input
•Log real errors server-side; return safe messages client-side

Gotchas

•Never expose stack traces or SQL errors to clients in production
•Catching Exception globally prevents Flask from returning HTML error pages for API routes
•request.get_json() returns None if Content-Type is not application/json

Use custom exception classes with proper status codes, a consistent error JSON format, and Flask error handlers. Validate input and return field-level errors. Never expose internal details in responses.

•Custom exceptions + error handlers for consistent JSON errors
•Log real errors server-side, return safe messages to clients

class APIError(Exception):
    def __init__(self, message, status_code=400):
        self.message = message
        self.status_code = status_code

@app.errorhandler(APIError)
def handle_api_error(error):
    return jsonify({'error': {'message': error.message}}), error.status_code

@app.errorhandler(Exception)
def handle_all(error):
    app.logger.exception('Unexpected error')
    return jsonify({'error': {'message': 'Internal error'}}), 500

Never expose stack traces or database errors in API responses

Question 14 / 18

What is web scraping with BeautifulSoup and what are the ethical considerations?

Web scraping is the automated extraction of data from websites by parsing HTML. BeautifulSoup is a Python library that makes parsing HTML/XML easy.

Key BeautifulSoup methods:

soup.find('tag') - Find the first matching element
soup.find_all('tag') - Find all matching elements
soup.select('css-selector') - CSS selector support
.text or .get_text() - Extract text content
.attrs or ['attribute'] - Access HTML attributes

Ethical and legal considerations:

Check robots.txt for scraping rules
Respect rate limits; add delays between requests
Check the site's Terms of Service
Use the site's API if one exists
Do not scrape personal data without consent (GDPR, CCPA)
Identify yourself with a proper User-Agent

Pseudo Code

1. Fetch the HTML page with requests.get()
2. Parse with BeautifulSoup(html, 'html.parser')
3. Navigate the DOM with find(), find_all(), select()
4. Extract text with .text or attributes with ['attr']
5. Handle pagination, errors, and rate limiting

Code

import requests
from bs4 import BeautifulSoup
import time

# Basic scraping
response = requests.get(
    'https://example.com',
    headers={'User-Agent': 'MyBot/1.0 (contact@mysite.com)'},
    timeout=10
)
soup = BeautifulSoup(response.text, 'html.parser')

# Find elements
title = soup.find('h1').text
print(title)

# Find all links
links = soup.find_all('a')
for link in links:
    href = link.get('href')  # or link['href']
    text = link.get_text(strip=True)
    print(f'{text}: {href}')

# CSS selectors
articles = soup.select('div.article > h2.title')
for article in articles:
    print(article.text)

# Find by attributes
email_input = soup.find('input', {'type': 'email', 'name': 'email'})
all_images = soup.find_all('img', class_='thumbnail')

# Navigate the tree
parent = soup.find('div', class_='content')
children = parent.find_all('p')  # Only direct and nested <p> tags
first_child = parent.find('p')    # First <p>

# Extract structured data
def scrape_products(url):
    response = requests.get(url, timeout=10)
    soup = BeautifulSoup(response.text, 'html.parser')

    products = []
    for card in soup.select('.product-card'):
        product = {
            'name': card.select_one('.product-name').get_text(strip=True),
            'price': card.select_one('.price').get_text(strip=True),
            'url': card.select_one('a')['href'],
        }
        products.append(product)
    return products

# Check robots.txt
def check_robots_txt(base_url, path):
    from urllib.robotparser import RobotFileParser
    rp = RobotFileParser()
    rp.set_url(f'{base_url}/robots.txt')
    rp.read()
    return rp.can_fetch('*', f'{base_url}{path}')

if check_robots_txt('https://example.com', '/products'):
    products = scrape_products('https://example.com/products')

# Polite scraping with delays
def scrape_with_delay(urls, delay=1):
    results = []
    for url in urls:
        response = requests.get(url, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results.append(soup.title.text if soup.title else 'No title')
        time.sleep(delay)  # Be polite
    return results

Key Points

•BeautifulSoup parses HTML; use find(), find_all(), and select() to navigate
•.text or .get_text(strip=True) extracts text; ['attr'] or .get('attr') for attributes
•select() supports CSS selectors (div.class > h2, #id, etc.)
•Always check robots.txt and respect rate limits when scraping
•Use the site's API if available instead of scraping

Gotchas

•Dynamic content loaded by JavaScript will not appear in requests.get(); use Selenium or Playwright for JS-heavy sites
•Websites change their HTML structure without notice; scrapers break frequently
•Scraping without rate limiting can get your IP blocked
•Some content behind login walls requires session cookies; use requests.Session()

BeautifulSoup parses HTML for data extraction. Use soup.find() / soup.select() to locate elements, .text for content, ['href'] for attributes. Always check robots.txt, add delays, and prefer APIs over scraping.

•find/find_all for tags, select for CSS selectors, .text for content
•Check robots.txt, add delays, prefer APIs

from bs4 import BeautifulSoup
import requests

soup = BeautifulSoup(requests.get(url).text, 'html.parser')
title = soup.find('h1').text
links = [(a.text, a['href']) for a in soup.find_all('a')]
prices = [p.text for p in soup.select('.product .price')]

JavaScript-rendered content requires Selenium/Playwright, not requests

Question 15 / 18

How do you implement rate limiting for a REST API?

Rate limiting restricts how many requests a client can make in a time window, protecting your API from abuse and overload.

Common algorithms:

Fixed window: Count requests in fixed time intervals (e.g., 100 per minute).
Sliding window: More accurate; considers a rolling time window.
Token bucket: Tokens are added at a fixed rate; each request costs a token.
Leaky bucket: Requests are processed at a steady rate; excess requests queue or are rejected.

Implementation approaches:

In-memory: Simple dict/counter (single server only).
Redis: Distributed rate limiting across multiple servers.
Flask-Limiter: Extension that handles everything.

Response headers for rate limits:

X-RateLimit-Limit: Max requests allowed
X-RateLimit-Remaining: Requests remaining
X-RateLimit-Reset: When the limit resets (Unix timestamp)
Retry-After: How long to wait (returned with 429)

Pseudo Code

1. Track request count per client (by IP or API key)
2. On each request, check if limit is exceeded
3. If exceeded, return 429 Too Many Requests
4. Include rate limit headers in all responses
5. Reset counters at the interval boundary

Code

# Method 1: Flask-Limiter (recommended)
from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)
limiter = Limiter(
    app=app,
    key_func=get_remote_address,  # Rate limit by IP
    default_limits=['200 per day', '50 per hour']
)

@app.route('/api/users')
@limiter.limit('10 per minute')  # Override default for this endpoint
def list_users():
    return jsonify({'users': []})

@app.route('/api/search')
@limiter.limit('5 per minute')  # Expensive endpoint
def search():
    return jsonify({'results': []})

@app.route('/api/health')
@limiter.exempt  # No rate limiting
def health():
    return jsonify({'status': 'ok'})

# Method 2: Manual rate limiting with in-memory store
import time
from collections import defaultdict
from flask import Flask, request, jsonify
from functools import wraps

rate_limit_store = defaultdict(list)

def rate_limit(max_requests, window_seconds):
    def decorator(f):
        @wraps(f)
        def wrapped(*args, **kwargs):
            client_ip = request.remote_addr
            now = time.time()
            key = f'{client_ip}:{f.__name__}'

            # Remove old entries outside the window
            rate_limit_store[key] = [
                t for t in rate_limit_store[key]
                if t > now - window_seconds
            ]

            if len(rate_limit_store[key]) >= max_requests:
                retry_after = int(window_seconds - (now - rate_limit_store[key][0]))
                response = jsonify({
                    'error': 'Rate limit exceeded',
                    'retry_after': retry_after
                })
                response.status_code = 429
                response.headers['Retry-After'] = str(retry_after)
                response.headers['X-RateLimit-Limit'] = str(max_requests)
                response.headers['X-RateLimit-Remaining'] = '0'
                return response

            rate_limit_store[key].append(now)

            response = f(*args, **kwargs)
            remaining = max_requests - len(rate_limit_store[key])
            response.headers['X-RateLimit-Limit'] = str(max_requests)
            response.headers['X-RateLimit-Remaining'] = str(remaining)
            return response
        return wrapped
    return decorator

@app.route('/api/data')
@rate_limit(max_requests=10, window_seconds=60)
def get_data():
    return jsonify({'data': 'value'})

Key Points

•Flask-Limiter is the easiest approach for Flask APIs
•Rate limit by IP, API key, or user ID depending on your auth model
•Return 429 Too Many Requests with Retry-After header when limit is exceeded
•Include X-RateLimit-* headers in all responses so clients can track usage
•Use Redis as the backend for distributed rate limiting across multiple servers

Gotchas

•In-memory rate limiting does not work with multiple server processes/workers
•Rate limiting by IP can block multiple users behind a shared IP (corporate networks, VPNs)
•Authenticated endpoints should rate limit by user/API key, not just IP

Rate limiting restricts requests per time window. Use Flask-Limiter for easy setup. Return 429 Too Many Requests with Retry-After header when exceeded. Rate limit by IP or API key. Use Redis for multi-server deployments.

•Flask-Limiter handles rate limiting with decorators
•Return 429 with Retry-After; use Redis for distributed setups

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(app=app, key_func=get_remote_address,
                  default_limits=['200 per day', '50 per hour'])

@app.route('/api/search')
@limiter.limit('5 per minute')
def search():
    return jsonify({'results': []})

In-memory rate limiting fails with multiple workers; use Redis

Question 16 / 18

How do you handle file uploads and downloads in REST APIs?

File uploads use multipart/form-data encoding, where the request body contains both file data and optional form fields. Downloads return file content with appropriate Content-Type and Content-Disposition headers.

Upload considerations:

Validate file type (extension and MIME type)
Limit file size
Generate safe filenames (avoid path traversal)
Store files outside the web root or in cloud storage

Download approaches:

Small files: Return directly from memory
Large files: Stream the response to avoid loading the entire file into memory

Pseudo Code

Upload:
1. Client sends POST with multipart/form-data
2. Server receives file via request.files
3. Validate type, size, and filename
4. Save to disk or cloud storage
5. Return file metadata (URL, id, size)

Download:
1. Client sends GET to the file endpoint
2. Server reads file from storage
3. Return with Content-Type and Content-Disposition headers
4. For large files, use streaming response

Code

import os
from flask import Flask, request, jsonify, send_file, send_from_directory
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = '/tmp/uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max

ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

# File upload endpoint
@app.route('/api/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return jsonify({'error': 'No file provided'}), 400

    file = request.files['file']
    if file.filename == '':
        return jsonify({'error': 'No file selected'}), 400

    if not allowed_file(file.filename):
        return jsonify({'error': 'File type not allowed'}), 400

    # Secure the filename to prevent path traversal
    filename = secure_filename(file.filename)

    # Save to upload folder
    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
    file.save(filepath)

    return jsonify({
        'filename': filename,
        'size': os.path.getsize(filepath),
        'url': f'/api/files/{filename}'
    }), 201

# Multiple file upload
@app.route('/api/upload-multiple', methods=['POST'])
def upload_multiple():
    files = request.files.getlist('files')
    uploaded = []
    for file in files:
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            uploaded.append(filename)
    return jsonify({'uploaded': uploaded}), 201

# File download endpoint
@app.route('/api/files/<filename>', methods=['GET'])
def download_file(filename):
    filename = secure_filename(filename)
    return send_from_directory(
        app.config['UPLOAD_FOLDER'],
        filename,
        as_attachment=True  # Forces download instead of display
    )

# Streaming large file download
@app.route('/api/export', methods=['GET'])
def export_large_file():
    def generate():
        # Generate CSV data line by line
        yield 'id,name,email\n'
        for i in range(100000):
            yield f'{i},User {i},user{i}@example.com\n'

    return app.response_class(
        generate(),
        mimetype='text/csv',
        headers={'Content-Disposition': 'attachment; filename=export.csv'}
    )

# Client-side upload with requests
import requests

# Single file upload
with open('photo.jpg', 'rb') as f:
    response = requests.post(
        'http://localhost:5000/api/upload',
        files={'file': ('photo.jpg', f, 'image/jpeg')}
    )

# Upload with additional form data
with open('document.pdf', 'rb') as f:
    response = requests.post(
        'http://localhost:5000/api/upload',
        files={'file': ('document.pdf', f, 'application/pdf')},
        data={'description': 'My document', 'category': 'reports'}
    )

Key Points

•Use multipart/form-data for uploads; request.files for Flask server-side
•Always validate file type, size, and use secure_filename() to prevent path traversal
•MAX_CONTENT_LENGTH limits upload size; returns 413 if exceeded
•send_from_directory() safely serves files; as_attachment=True forces download
•Stream large downloads with generator functions to avoid memory issues

Gotchas

•secure_filename() may return empty string for filenames in non-Latin characters
•Storing uploads in the web root is a security risk; use a separate directory
•MAX_CONTENT_LENGTH rejection happens before your view function runs

Use request.files for uploads with secure_filename() for safety and send_from_directory() for downloads. Validate file type and size. Set MAX_CONTENT_LENGTH in Flask config. Stream large files with generators.

•secure_filename() prevents path traversal; validate file type and size
•send_from_directory() for safe downloads; stream large files

from werkzeug.utils import secure_filename

@app.route('/api/upload', methods=['POST'])
def upload():
    file = request.files['file']
    filename = secure_filename(file.filename)
    file.save(f'/tmp/uploads/{filename}')
    return jsonify({'filename': filename}), 201

@app.route('/api/files/<name>')
def download(name):
    return send_from_directory('/tmp/uploads', secure_filename(name))

Never store uploads in the web root; use a separate directory

Question 17 / 18

How do you make asynchronous HTTP requests in Python?

When you need to make many HTTP requests, synchronous requests library is slow because it waits for each response before starting the next. Asynchronous HTTP clients allow concurrent requests.

Options:

aiohttp - Async HTTP client/server built on asyncio. Most mature.
httpx - Modern library supporting both sync and async. Drop-in replacement for requests.
concurrent.futures - Run synchronous requests in a thread pool.

httpx is recommended for new projects because its API is almost identical to requests, making migration easy.

Pseudo Code

1. Choose an async HTTP client (httpx or aiohttp)
2. Define async functions with async def
3. Use asyncio.gather() to run requests concurrently
4. Await each response
5. Process all results when gather completes

Code

import asyncio
import httpx
import aiohttp
import time

# Method 1: httpx (recommended - closest to requests API)
async def fetch_with_httpx():
    async with httpx.AsyncClient() as client:
        # Single request
        response = await client.get('https://httpbin.org/get', timeout=10)
        print(response.json())

        # Multiple concurrent requests
        urls = [f'https://httpbin.org/get?id={i}' for i in range(10)]
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        results = [r.json() for r in responses]
        return results

asyncio.run(fetch_with_httpx())

# Method 2: aiohttp
async def fetch_with_aiohttp():
    async with aiohttp.ClientSession() as session:
        # Single request
        async with session.get('https://httpbin.org/get') as response:
            data = await response.json()
            print(data)

        # Multiple concurrent requests
        async def fetch_one(session, url):
            async with session.get(url) as response:
                return await response.json()

        urls = [f'https://httpbin.org/get?id={i}' for i in range(10)]
        tasks = [fetch_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

asyncio.run(fetch_with_aiohttp())

# Method 3: concurrent.futures with synchronous requests
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_sync(url):
    return requests.get(url, timeout=10).json()

def fetch_all_threaded(urls, max_workers=10):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_url = {executor.submit(fetch_sync, url): url for url in urls}
        for future in as_completed(future_to_url):
            try:
                results.append(future.result())
            except Exception as e:
                print(f'Error: {e}')
    return results

urls = [f'https://httpbin.org/get?id={i}' for i in range(10)]
results = fetch_all_threaded(urls)

# Performance comparison
def benchmark():
    urls = [f'https://httpbin.org/delay/1' for _ in range(5)]

    # Synchronous: ~5 seconds
    start = time.time()
    for url in urls:
        requests.get(url, timeout=10)
    print(f'Sync: {time.time() - start:.1f}s')

    # Async: ~1 second
    async def async_fetch():
        async with httpx.AsyncClient() as client:
            tasks = [client.get(url) for url in urls]
            await asyncio.gather(*tasks)

    start = time.time()
    asyncio.run(async_fetch())
    print(f'Async: {time.time() - start:.1f}s')

Key Points

•httpx is the recommended modern async client (sync and async in one library)
•asyncio.gather() runs multiple requests concurrently
•Always use async context managers (async with) for client sessions
•ThreadPoolExecutor is an easy way to parallelize synchronous requests
•10 concurrent 1-second requests take ~1 second async vs ~10 seconds sync

Gotchas

•asyncio.run() cannot be called inside an already-running event loop (common in Jupyter)
•aiohttp response.json() is a coroutine (needs await); requests response.json() is not
•Too many concurrent requests can overwhelm the server; use semaphores to limit concurrency

Use httpx.AsyncClient with asyncio.gather() for concurrent requests. Alternative: concurrent.futures.ThreadPoolExecutor for parallelizing synchronous requests. Async is much faster for many requests (5x with 1s latency each).

•httpx for async requests; asyncio.gather for concurrency
•ThreadPoolExecutor for parallelizing synchronous requests

import asyncio
import httpx

async def fetch_all(urls):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

results = asyncio.run(fetch_all(['https://api.example.com/1', 'https://api.example.com/2']))

Limit concurrency with semaphores to avoid overwhelming the server

Question 18 / 18

How do you design and consume RESTful API endpoints that support filtering, sorting, and searching?

A well-designed REST API supports flexible data retrieval through query parameters for filtering, sorting, and searching.

Filtering: Narrow results by field values.

Equality: ?status=active
Comparison: ?price_min=10&price_max=100 or ?price[gte]=10&price[lte]=100
Multiple values: ?category=books,electronics or ?category=books&category=electronics

Sorting: Order results by one or more fields.

?sort=name (ascending by default)
?sort=-created_at (prefix - for descending)
?sort=name,-created_at (multi-field)

Searching: Full-text or partial match.

?search=python (global search across relevant fields)
?name__contains=alice (field-specific)

Pseudo Code

1. Parse query parameters from the request
2. Build database query dynamically based on filters
3. Apply sorting with order_by
4. Apply pagination with limit/offset
5. Return results with metadata about applied filters

Code

from flask import Flask, request, jsonify
from sqlalchemy import asc, desc
from sqlalchemy.orm import Session

app = Flask(__name__)

@app.route('/api/products', methods=['GET'])
def list_products():
    # Parse query parameters
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    sort = request.args.get('sort', 'name')
    search = request.args.get('search', '')

    # Filters
    category = request.args.get('category')
    min_price = request.args.get('min_price', type=float)
    max_price = request.args.get('max_price', type=float)
    is_active = request.args.get('is_active', type=lambda v: v.lower() == 'true')

    with Session(engine) as session:
        query = session.query(Product)

        # Apply filters
        if category:
            categories = category.split(',')
            query = query.filter(Product.category.in_(categories))
        if min_price is not None:
            query = query.filter(Product.price >= min_price)
        if max_price is not None:
            query = query.filter(Product.price <= max_price)
        if is_active is not None:
            query = query.filter(Product.is_active == is_active)

        # Apply search
        if search:
            search_filter = f'%{search}%'
            query = query.filter(
                Product.name.ilike(search_filter) |
                Product.description.ilike(search_filter)
            )

        # Apply sorting
        for sort_field in sort.split(','):
            sort_field = sort_field.strip()
            if sort_field.startswith('-'):
                column = getattr(Product, sort_field[1:], None)
                if column:
                    query = query.order_by(desc(column))
            else:
                column = getattr(Product, sort_field, None)
                if column:
                    query = query.order_by(asc(column))

        # Get total count before pagination
        total = query.count()

        # Apply pagination
        products = query.offset((page - 1) * per_page).limit(per_page).all()

        return jsonify({
            'data': [{'id': p.id, 'name': p.name, 'price': p.price} for p in products],
            'meta': {
                'page': page,
                'per_page': per_page,
                'total': total,
                'total_pages': (total + per_page - 1) // per_page,
            },
            'filters_applied': {
                'category': category,
                'min_price': min_price,
                'max_price': max_price,
                'search': search,
                'sort': sort,
            }
        })

# Client consuming the API
import requests

response = requests.get(
    'https://api.example.com/products',
    params={
        'category': 'books,electronics',
        'min_price': 10,
        'max_price': 100,
        'sort': '-price,name',
        'search': 'python',
        'page': 1,
        'per_page': 25,
    },
    timeout=10
)
data = response.json()
print(f"Found {data['meta']['total']} products")
for product in data['data']:
    print(f"  {product['name']}: ${product['price']}")

Key Points

•Use query parameters for filtering (?field=value), sorting (?sort=-field), and searching (?search=term)
•Prefix with - for descending sort; comma-separate for multi-field sorting
•Validate and whitelist sortable/filterable fields to prevent SQL injection via getattr
•Return metadata about applied filters so clients can verify their request
•Always paginate list endpoints to avoid returning unbounded result sets

Gotchas

•Using getattr with user input for column names can expose internal model attributes; whitelist allowed fields
•Counting total results (query.count()) can be slow on large tables; consider caching or approximate counts
•Multiple values for the same param (?tag=a&tag=b) are handled with request.args.getlist('tag') in Flask

Use query params for filtering (?status=active), sorting (?sort=-created_at), and searching (?search=term). Build the database query dynamically. Always paginate and whitelist allowed filter/sort fields.

•Query params for filtering/sorting/search; dynamic query building
•Whitelist allowed fields; always paginate

# API: GET /api/products?category=books&min_price=10&sort=-price&search=python&page=1

query = session.query(Product)
if category:
    query = query.filter(Product.category == category)
if search:
    query = query.filter(Product.name.ilike(f'%{search}%'))
if sort.startswith('-'):
    query = query.order_by(desc(getattr(Product, sort[1:])))
products = query.offset((page-1)*per_page).limit(per_page).all()

Whitelist sortable/filterable fields to prevent information leakage

Built by onllm.dev

Home GitHub Support awesomePrep