45.1 Introduction to APIs
API means Application Programming Interface. An API allows one program to communicate with another program through a defined set of rules. A weather application may request forecast information from a weather API. A shopping application may send orders to a payment API. APIs allow applications to exchange information without giving direct access to their internal code or database.
Example: Simulate an API function
# Create a function that behaves like a simple API.
def get_product(product_id):
# Store example product information.
products = {
1: {
"id": 1,
"name": "Keyboard",
"price": 49.99
},
2: {
"id": 2,
"name": "Mouse",
"price": 24.50
}
}
# Return the requested product.
return products.get(product_id)
# Request product number 1.
response = get_product(1)
print(response)
Output:
{'id': 1, 'name': 'Keyboard', 'price': 49.99}
Output Explanation:
The calling program provides a product ID, and the API-like function returns related product information.
45.2 REST APIs
REST means Representational State Transfer. A REST API normally organizes information into resources such as users, products, courses, and orders. Each resource is identified by a URL. HTTP methods describe the requested operation. REST APIs commonly exchange JSON and use standard HTTP status codes to describe successful and unsuccessful requests.
Example: REST-style resources
# Store example REST endpoints and their meanings.
endpoints = {
"GET /products": "List products",
"GET /products/1": "Get product 1",
"POST /products": "Create a product",
"PATCH /products/1": "Update product 1",
"DELETE /products/1": "Delete product 1"
}
for endpoint, meaning in endpoints.items():
print(endpoint, "-", meaning)
Output:
GET /products - List products
GET /products/1 - Get product 1
POST /products - Create a product
PATCH /products/1 - Update product 1
DELETE /products/1 - Delete product 1
Output Explanation:
The same products resource supports several operations. The HTTP method and URL together describe the action.
45.3 API Endpoints
An API endpoint is a specific URL where an API accepts requests. A base URL identifies the API service, while the endpoint path identifies a resource or action. For example, an API may use https://api.example.com as its base URL and /products as a product endpoint.
Example: Build endpoint URLs
# Store the API base URL.
base_url = "https://api.example.com"
# Store resource paths.
products_path = "/products"
users_path = "/users"
# Combine the base URL and paths.
products_url = base_url + products_path
users_url = base_url + users_path
print(products_url)
print(users_url)
Output:
https://api.example.com/products
https://api.example.com/users
Output Explanation:
Each complete endpoint combines the API’s base address with a specific resource path.
45.4 HTTP Methods
APIs use HTTP methods to describe operations. GET retrieves information. POST usually creates a resource. PUT usually replaces a complete resource. PATCH usually changes selected fields. DELETE removes a resource. APIs should document which methods are available for every endpoint.
Example: Select an HTTP method
# Store an application action.
action = "create"
# Match the action to an HTTP method.
method_map = {
"read": "GET",
"create": "POST",
"replace": "PUT",
"update": "PATCH",
"delete": "DELETE"
}
method = method_map[action]
print("Action:", action)
print("HTTP method:", method)
Output:
Action: create
HTTP method: POST
Output Explanation:
Creating a new resource is commonly represented by a POST request.
45.5 Query Parameters
Query parameters add optional information to a URL. They appear after a question mark and are separated by ampersands. APIs use query parameters for filtering, sorting, searching, pagination, and optional settings. The requests library can build query strings safely from a Python dictionary.
Example: Create query parameters
# Import URL encoding support.
from urllib.parse import urlencode
# Create query parameters.
parameters = {
"category": "books",
"page": 2,
"limit": 10
}
# Convert parameters to a query string.
query_string = urlencode(parameters)
url = (
"https://api.example.com/products?"
+ query_string
)
print(url)
Output:
https://api.example.com/products?category=books&page=2&limit=10
Output Explanation:
The query string tells the API to return page two, limit the result to ten items, and filter products by the books category.
45.6 Path Parameters
A path parameter is a value placed directly inside an endpoint path. It normally identifies a specific resource. In /products/15, the value 15 identifies one product. Path parameters are usually required, while query parameters are often optional.
Example: Build a URL with a path parameter
# Store the product identifier.
product_id = 15
# Insert the ID into the endpoint path.
url = (
f"https://api.example.com/"
f"products/{product_id}"
)
print(url)
Output:
https://api.example.com/products/15
Output Explanation:
The endpoint points to product number 15 rather than the complete products collection.
45.7 Request Headers
Request headers provide extra information about an API request. They may describe accepted response formats, authentication credentials, language preferences, content types, and client information. Headers are name-value pairs. Authentication headers must be protected because they may contain secrets.
Example: Create API request headers
# Create a dictionary of request headers.
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "PythonCourseClient/1.0"
}
for name, value in headers.items():
print(f"{name}: {value}")
Output:
Accept: application/json
Content-Type: application/json
User-Agent: PythonCourseClient/1.0
Output Explanation:
The client requests JSON, says that its request body is JSON, and identifies the client application.
45.8 Request Bodies
A request body contains information sent to an API. POST, PUT, and PATCH requests often include bodies. JSON is a common body format. The body may contain a new product, account information, search settings, or fields that need updating. APIs should validate every received field.
Example: Create a JSON request body
# Import JSON support.
import json
# Create a Python dictionary.
product_data = {
"name": "Keyboard",
"price": 49.99,
"quantity": 10
}
# Convert the dictionary to JSON text.
json_body = json.dumps(
product_data,
indent=2
)
print(json_body)
Output:
{
"name": "Keyboard",
"price": 49.99,
"quantity": 10
}
Output Explanation:
The Python dictionary is converted into JSON text that can be transmitted in an HTTP request body.
45.9 JSON Responses
APIs commonly return JSON responses. JSON supports objects, arrays, strings, numbers, Boolean values, and null. Python usually converts JSON objects into dictionaries and JSON arrays into lists. Applications should confirm that the response really contains JSON before depending on its structure.
Example: Parse a JSON response
# Import JSON support.
import json
# Store JSON returned by an API.
response_text = """
{
"id": 1,
"name": "Keyboard",
"price": 49.99,
"available": true
}
"""
# Convert JSON text to a dictionary.
product = json.loads(response_text)
print(product["name"])
print(product["price"])
print(product["available"])
Output:
Keyboard
49.99
True
Output Explanation:
JSON text is converted into a Python dictionary. The JSON value true becomes the Python value True.
45.10 The requests Library
The requests library provides a convenient way to send HTTP requests from Python. It supports methods such as GET, POST, PUT, PATCH, and DELETE. It also manages query parameters, headers, JSON bodies, cookies, authentication, sessions, timeouts, and response information.
Example: Install requests
python -m pip install requests
Example: Inspect the installed version
# Import the requests library.
import requests
print("requests version:", requests.__version__)
Example Output:
requests version: 2.32.3
Output Explanation:
The exact version depends on the version installed in the current Python environment.
45.11 GET Requests
A GET request retrieves information from an API. Query parameters can filter or limit the returned information. The response object contains the status code, headers, body text, URL, and other details. A timeout should normally be included to prevent the application from waiting forever.
Example: Send a GET request
# Import requests.
import requests
# Send a request to a public testing API.
response = requests.get(
"https://jsonplaceholder.typicode.com/posts/1",
timeout=10
)
# Raise an exception for unsuccessful responses.
response.raise_for_status()
# Convert the JSON response to a dictionary.
post = response.json()
print("Status:", response.status_code)
print("Post ID:", post["id"])
print("Title:", post["title"])
Example Output:
Status: 200
Post ID: 1
Title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit
Output Explanation:
The API returns status 200 and a JSON object representing post number one.
Example: GET request with query parameters
# Store query parameters.
parameters = {
"userId": 1
}
# Send the parameters safely.
response = requests.get(
"https://jsonplaceholder.typicode.com/posts",
params=parameters,
timeout=10
)
response.raise_for_status()
posts = response.json()
print("Final URL:", response.url)
print("Posts returned:", len(posts))
print("First post ID:", posts[0]["id"])
Example Output:
Final URL: https://jsonplaceholder.typicode.com/posts?userId=1
Posts returned: 10
First post ID: 1
Output Explanation:
Requests converts the parameters dictionary into a correctly encoded query string.
45.12 POST Requests
A POST request commonly creates a new resource or submits information for processing. The requests library can convert a Python dictionary into JSON by using the json argument. It automatically serializes the dictionary and adds the appropriate JSON content-type header.
Example: Create a resource with POST
# Import requests.
import requests
# Create the new post information.
new_post = {
"title": "Learning APIs",
"body": "Python can send HTTP requests.",
"userId": 1
}
# Send the JSON body.
response = requests.post(
"https://jsonplaceholder.typicode.com/posts",
json=new_post,
timeout=10
)
response.raise_for_status()
created_post = response.json()
print("Status:", response.status_code)
print("Created ID:", created_post["id"])
print("Title:", created_post["title"])
Example Output:
Status: 201
Created ID: 101
Title: Learning APIs
Output Explanation:
Status 201 means the API reports that a new resource was created. This testing API simulates creation without permanently storing the new post.
45.13 PUT and PATCH Requests
PUT commonly replaces the complete representation of a resource. PATCH commonly changes only selected fields. The exact behavior depends on the API documentation. When using PUT, required fields usually need to be included. PATCH is often more convenient when only one or two values need modification.
Example: Replace a resource with PUT
# Import requests.
import requests
# Create a complete replacement object.
replacement_post = {
"id": 1,
"title": "Updated API Lesson",
"body": "This resource was replaced.",
"userId": 1
}
response = requests.put(
"https://jsonplaceholder.typicode.com/posts/1",
json=replacement_post,
timeout=10
)
response.raise_for_status()
result = response.json()
print("Status:", response.status_code)
print("Title:", result["title"])
print("Body:", result["body"])
Example Output:
Status: 200
Title: Updated API Lesson
Body: This resource was replaced.
Output Explanation:
The PUT request sends a complete replacement representation for post number one.
Example: Update one field with PATCH
# Create only the field that needs changing.
partial_update = {
"title": "Partially Updated Title"
}
response = requests.patch(
"https://jsonplaceholder.typicode.com/posts/1",
json=partial_update,
timeout=10
)
response.raise_for_status()
result = response.json()
print("Status:", response.status_code)
print("Title:", result["title"])
Example Output:
Status: 200
Title: Partially Updated Title
Output Explanation:
The PATCH body contains only the title field, indicating a partial change.
45.14 DELETE Requests
A DELETE request asks an API to remove a resource. Some APIs return a JSON confirmation, while others return status 204 with no response body. A successful DELETE request should not automatically be repeated unless the operation is known to be safe and idempotent.
Example: Delete a resource
# Import requests.
import requests
response = requests.delete(
"https://jsonplaceholder.typicode.com/posts/1",
timeout=10
)
response.raise_for_status()
print("Status:", response.status_code)
print("Request successful:", response.ok)
Example Output:
Status: 200
Request successful: True
Output Explanation:
The testing API reports a successful deletion. Real APIs may instead return status 204 with an empty body.
45.15 Authentication
Authentication verifies the identity of a client or user. APIs may use usernames and passwords, API keys, bearer tokens, signed requests, certificates, or OAuth. Authentication information should be transmitted only through HTTPS and should never be placed directly in public source code.
Example: Basic authentication
# Import requests.
import requests
# Send HTTP Basic Authentication credentials.
response = requests.get(
"https://httpbin.org/basic-auth/student/python",
auth=(
"student",
"python"
),
timeout=10
)
response.raise_for_status()
print(response.json())
Example Output:
{'authenticated': True, 'user': 'student'}
Output Explanation:
The server confirms that the supplied username and password were accepted. Basic authentication should be used only through HTTPS.
45.16 API Keys
An API key is a secret value that identifies an application or account. APIs may expect the key in a header or query parameter. Keys should be loaded from environment variables or a secure secret manager. They should not be committed to Git, printed in logs, or shared in screenshots.
Example: Load an API key safely
# Import operating-system support.
import os
# Read the key from an environment variable.
api_key = os.getenv("EXAMPLE_API_KEY")
if not api_key:
raise ValueError(
"EXAMPLE_API_KEY is required."
)
# Add the key to a request header.
headers = {
"X-API-Key": api_key
}
print("API key loaded:", bool(api_key))
print("Header prepared:", "X-API-Key" in headers)
Output:
API key loaded: True
Header prepared: True
Output Explanation:
The program confirms that the key exists without displaying the secret value.
45.17 Bearer Tokens
A bearer token is a credential commonly placed in the Authorization header. The word Bearer is followed by the token value. Anyone who obtains a valid bearer token may be able to use it, so the token must be protected. Tokens should be sent only through HTTPS and should usually have limited permissions and expiration times.
Example: Create a bearer token header
# Import environment-variable support.
import os
token = os.getenv("API_ACCESS_TOKEN")
if not token:
raise ValueError(
"API_ACCESS_TOKEN is required."
)
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/json"
}
print("Authorization header created.")
print("Token displayed: No")
Output:
Authorization header created.
Token displayed: No
Output Explanation:
The token is included in the header but is not printed to the terminal.
45.18 OAuth Concepts
OAuth is an authorization framework that allows a user to grant an application limited access without sharing the user’s password with that application. OAuth commonly involves a resource owner, client application, authorization server, and resource server. After approval, the client receives an access token that can be used to call protected APIs.
Example: Simplified OAuth flow
# Store simplified OAuth steps.
oauth_steps = [
"1. Client requests authorization",
"2. User approves requested permissions",
"3. Authorization server returns a code",
"4. Client exchanges code for access token",
"5. Client calls API with access token"
]
for step in oauth_steps:
print(step)
Output:
1. Client requests authorization
2. User approves requested permissions
3. Authorization server returns a code
4. Client exchanges code for access token
5. Client calls API with access token
Output Explanation:
The client receives permission through the authorization server rather than directly collecting the user’s password.
45.19 Pagination
Pagination divides a large result set into smaller pages. APIs may use page numbers, offsets, cursors, or continuation tokens. Pagination reduces response size and memory usage. A client must continue requesting pages until the API indicates that no more results remain.
Example: Page-number pagination
# Import requests.
import requests
all_posts = []
page = 1
page_size = 20
while True:
response = requests.get(
"https://jsonplaceholder.typicode.com/posts",
params={
"_page": page,
"_limit": page_size
},
timeout=10
)
response.raise_for_status()
current_page = response.json()
# Stop when the API returns no items.
if not current_page:
break
all_posts.extend(current_page)
print(
f"Page {page}: "
f"{len(current_page)} posts"
)
page += 1
print("Total posts:", len(all_posts))
Example Output:
Page 1: 20 posts
Page 2: 20 posts
Page 3: 20 posts
Page 4: 20 posts
Page 5: 20 posts
Total posts: 100
Output Explanation:
The client requests twenty posts at a time and combines the pages into one list.
45.20 Rate Limits
A rate limit controls how many requests a client may send during a time period. APIs use rate limits to protect availability and prevent abuse. Rate-limit information may appear in response headers. When a client exceeds the limit, the API commonly returns status 429. The client should wait before trying again.
Example: Inspect rate-limit headers
# Simulate rate-limit response headers.
headers = {
"X-RateLimit-Limit": "100",
"X-RateLimit-Remaining": "42",
"X-RateLimit-Reset": "60"
}
limit = int(
headers["X-RateLimit-Limit"]
)
remaining = int(
headers["X-RateLimit-Remaining"]
)
reset_seconds = int(
headers["X-RateLimit-Reset"]
)
print("Request limit:", limit)
print("Requests remaining:", remaining)
print("Reset in seconds:", reset_seconds)
Output:
Request limit: 100
Requests remaining: 42
Reset in seconds: 60
Output Explanation:
The client has used 58 of its 100 available requests. The limit resets in 60 seconds.
Example: Handle status 429
# Import time and requests.
import time
import requests
response = requests.get(
"https://api.example.com/data",
timeout=10
)
if response.status_code == 429:
# Read the suggested wait time.
retry_after = int(
response.headers.get(
"Retry-After",
"60"
)
)
print(
f"Rate limited. Wait "
f"{retry_after} seconds."
)
time.sleep(retry_after)
Possible Output:
Rate limited. Wait 60 seconds.
Output Explanation:
The client reads the server’s Retry-After instruction and waits before making another request.
45.21 Timeouts
A timeout limits how long a client waits for a network operation. Without a timeout, a program may wait indefinitely when a server is unavailable or extremely slow. Requests supports separate connection and read timeouts. A timeout does not guarantee that the entire response completes within one total time period.
Example: Use connection and read timeouts
# Import requests.
import requests
try:
response = requests.get(
"https://jsonplaceholder.typicode.com/posts/1",
# Wait up to 3 seconds to connect
# and 10 seconds for response data.
timeout=(3, 10)
)
response.raise_for_status()
print("Request completed.")
except requests.Timeout:
print("The request timed out.")
Possible Output:
Request completed.
Output Explanation:
The first timeout value controls connection waiting, while the second controls waiting for response data.
45.22 Retries
A retry repeats a failed request when the failure may be temporary. Suitable retry conditions may include connection errors, timeouts, status 429, or selected server errors. Retries should use a delay and a maximum attempt count. Automatically retrying operations that create or change data can cause duplicate actions unless the API supports idempotency.
Example: Configure retries
# Import requests and retry support.
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Configure retry behavior.
retry_policy = Retry(
total=3,
connect=3,
read=3,
# Retry selected temporary responses.
status_forcelist=[
429,
500,
502,
503,
504
],
# Retry only safe request methods.
allowed_methods=[
"GET",
"HEAD",
"OPTIONS"
],
# Increase delay between attempts.
backoff_factor=1,
# Respect Retry-After headers.
respect_retry_after_header=True
)
# Create a session.
session = requests.Session()
# Attach retry behavior to HTTPS requests.
session.mount(
"https://",
HTTPAdapter(
max_retries=retry_policy
)
)
response = session.get(
"https://jsonplaceholder.typicode.com/posts/1",
timeout=(3, 10)
)
response.raise_for_status()
print("Status:", response.status_code)
session.close()
Example Output:
Status: 200
Output Explanation:
The session can retry selected temporary failures. Successful requests return normally without using additional attempts.
45.23 API Error Handling
API calls can fail because of connection problems, invalid addresses, timeouts, authentication errors, rate limits, client mistakes, or server failures. Applications should catch specific request exceptions, inspect status codes, safely parse error responses, and display useful messages without exposing secrets.
Example: Complete request error handling
# Import requests.
import requests
def get_json(url):
try:
# Send the request with a timeout.
response = requests.get(
url,
timeout=(3, 10)
)
# Raise HTTPError for 4xx or 5xx.
response.raise_for_status()
try:
# Return parsed JSON.
return response.json()
except requests.JSONDecodeError:
print(
"The response was not valid JSON."
)
except requests.Timeout:
print("The request timed out.")
except requests.ConnectionError:
print(
"The API could not be reached."
)
except requests.HTTPError as error:
status_code = (
error.response.status_code
if error.response is not None
else "unknown"
)
print(
"The API returned status:",
status_code
)
except requests.RequestException as error:
print(
"The request failed:",
type(error).__name__
)
return None
data = get_json(
"https://jsonplaceholder.typicode.com/posts/1"
)
if data:
print("Title:", data["title"])
Example Output:
Title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit
Output Explanation:
The function handles network, HTTP, timeout, and JSON problems separately. Successful JSON is returned to the calling code.
45.24 Creating API Clients
An API client class places repeated request logic in one reusable location. It can manage the base URL, headers, authentication, sessions, timeouts, retries, error handling, and endpoint methods. This prevents application code from repeating the same low-level HTTP details throughout the project.
Example: Reusable API client
# Import requests.
import requests
class APIClient:
def __init__(
self,
base_url,
token=None,
timeout=(3, 10)
):
# Remove a trailing slash.
self.base_url = base_url.rstrip("/")
# Store timeout settings.
self.timeout = timeout
# Create a reusable session.
self.session = requests.Session()
# Set common headers.
self.session.headers.update(
{
"Accept": "application/json",
"User-Agent": (
"PythonCourseClient/1.0"
)
}
)
# Add authentication when supplied.
if token:
self.session.headers.update(
{
"Authorization": (
f"Bearer {token}"
)
}
)
def build_url(self, path):
# Ensure exactly one slash.
return (
f"{self.base_url}/"
f"{path.lstrip('/')}"
)
def get(self, path, params=None):
response = self.session.get(
self.build_url(path),
params=params,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
def post(self, path, data):
response = self.session.post(
self.build_url(path),
json=data,
timeout=self.timeout
)
response.raise_for_status()
return response.json()
def close(self):
self.session.close()
client = APIClient(
"https://jsonplaceholder.typicode.com"
)
try:
post = client.get("/posts/1")
print(post["id"])
print(post["title"])
finally:
client.close()
Example Output:
1
sunt aut facere repellat provident occaecati excepturi optio reprehenderit
Output Explanation:
The class manages the base URL, session, headers, timeout, request, status validation, and JSON conversion.
45.25 Chapter Project
In this project, you will create a command-line Post Manager that communicates with a REST API. The application can list posts, view one post, search posts by user, create a post, replace a post, partially update a post, and delete a post. The project demonstrates sessions, retries, timeouts, query parameters, path parameters, headers, JSON bodies, pagination, and error handling.
Project structure
api_post_manager/
│
├── app.py
├── api_client.py
├── config.py
└── post_service.py
Step 1: Install requests
python -m pip install requests
Step 2: Create config.py
# config.py
# Import environment-variable support.
import os
class Settings:
# Use a public testing API by default.
API_BASE_URL = os.getenv(
"API_BASE_URL",
"https://jsonplaceholder.typicode.com"
)
# Read an optional bearer token.
API_TOKEN = os.getenv(
"API_TOKEN"
)
# Store connection and read timeouts.
CONNECT_TIMEOUT = float(
os.getenv(
"CONNECT_TIMEOUT",
"3"
)
)
READ_TIMEOUT = float(
os.getenv(
"READ_TIMEOUT",
"10"
)
)
# Store pagination size.
PAGE_SIZE = int(
os.getenv(
"PAGE_SIZE",
"10"
)
)
def validate_settings():
if not Settings.API_BASE_URL.startswith(
("http://", "https://")
):
raise ValueError(
"API_BASE_URL must start with "
"http:// or https://"
)
if Settings.CONNECT_TIMEOUT <= 0:
raise ValueError(
"CONNECT_TIMEOUT must be positive."
)
if Settings.READ_TIMEOUT <= 0:
raise ValueError(
"READ_TIMEOUT must be positive."
)
if Settings.PAGE_SIZE <= 0:
raise ValueError(
"PAGE_SIZE must be positive."
)
Step 3: Create api_client.py
# api_client.py
# Import typing support.
from typing import Any
# Import requests.
import requests
# Import retry support.
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class APIClientError(Exception):
"""
Base exception for API client failures.
"""
class APIConnectionError(APIClientError):
"""
Raised when the API cannot be reached.
"""
class APITimeoutError(APIClientError):
"""
Raised when a request times out.
"""
class APIResponseError(APIClientError):
"""
Raised for unsuccessful HTTP responses.
"""
def __init__(
self,
status_code,
message,
response_data=None
):
super().__init__(message)
self.status_code = status_code
self.response_data = response_data
class APIClient:
def __init__(
self,
base_url,
token=None,
timeout=(3, 10)
):
# Normalize the base address.
self.base_url = base_url.rstrip("/")
# Store timeout settings.
self.timeout = timeout
# Create a persistent session.
self.session = requests.Session()
# Add common request headers.
self.session.headers.update(
{
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": (
"PythonPostManager/1.0"
)
}
)
# Add bearer authentication when supplied.
if token:
self.session.headers.update(
{
"Authorization": (
f"Bearer {token}"
)
}
)
# Configure safe retries.
retry_policy = Retry(
total=3,
connect=3,
read=3,
status_forcelist=[
429,
500,
502,
503,
504
],
allowed_methods=[
"GET",
"HEAD",
"OPTIONS"
],
backoff_factor=1,
respect_retry_after_header=True
)
adapter = HTTPAdapter(
max_retries=retry_policy
)
self.session.mount(
"https://",
adapter
)
self.session.mount(
"http://",
adapter
)
def build_url(self, path):
"""
Combine the base URL and endpoint path.
"""
return (
f"{self.base_url}/"
f"{path.lstrip('/')}"
)
def parse_response(self, response):
"""
Validate the status and return JSON.
"""
if not response.ok:
error_data = None
message = (
f"API request failed with "
f"status {response.status_code}."
)
try:
error_data = response.json()
if isinstance(error_data, dict):
message = (
error_data.get("message")
or error_data.get("error")
or message
)
except requests.JSONDecodeError:
if response.text.strip():
message = response.text.strip()[
:300
]
raise APIResponseError(
response.status_code,
message,
error_data
)
# Some successful responses have no body.
if response.status_code == 204:
return None
if not response.content:
return None
try:
return response.json()
except requests.JSONDecodeError as error:
raise APIResponseError(
response.status_code,
"The API returned invalid JSON."
) from error
def request(
self,
method,
path,
params=None,
json_data=None
) -> Any:
"""
Send one HTTP request.
"""
url = self.build_url(path)
try:
response = self.session.request(
method=method,
url=url,
params=params,
json=json_data,
timeout=self.timeout
)
return self.parse_response(
response
)
except requests.Timeout as error:
raise APITimeoutError(
"The API request timed out."
) from error
except requests.ConnectionError as error:
raise APIConnectionError(
"The API could not be reached."
) from error
except requests.RequestException as error:
raise APIClientError(
f"The request failed: "
f"{type(error).__name__}"
) from error
def get(self, path, params=None):
return self.request(
"GET",
path,
params=params
)
def post(self, path, data):
return self.request(
"POST",
path,
json_data=data
)
def put(self, path, data):
return self.request(
"PUT",
path,
json_data=data
)
def patch(self, path, data):
return self.request(
"PATCH",
path,
json_data=data
)
def delete(self, path):
return self.request(
"DELETE",
path
)
def close(self):
self.session.close()
Step 4: Create post_service.py
# post_service.py
class PostService:
def __init__(self, api_client):
# Store the reusable API client.
self.client = api_client
def list_posts(self, page=1, limit=10):
"""
Return one page of posts.
"""
return self.client.get(
"/posts",
params={
"_page": page,
"_limit": limit
}
)
def get_post(self, post_id):
"""
Return one post by ID.
"""
self.validate_id(post_id)
return self.client.get(
f"/posts/{post_id}"
)
def get_posts_by_user(self, user_id):
"""
Return posts written by one user.
"""
self.validate_id(user_id)
return self.client.get(
"/posts",
params={
"userId": user_id
}
)
def create_post(
self,
title,
body,
user_id
):
"""
Create a new post.
"""
title = self.validate_text(
title,
"title"
)
body = self.validate_text(
body,
"body"
)
self.validate_id(user_id)
return self.client.post(
"/posts",
{
"title": title,
"body": body,
"userId": user_id
}
)
def replace_post(
self,
post_id,
title,
body,
user_id
):
"""
Replace a complete post.
"""
self.validate_id(post_id)
self.validate_id(user_id)
title = self.validate_text(
title,
"title"
)
body = self.validate_text(
body,
"body"
)
return self.client.put(
f"/posts/{post_id}",
{
"id": post_id,
"title": title,
"body": body,
"userId": user_id
}
)
def update_title(
self,
post_id,
title
):
"""
Partially update a post title.
"""
self.validate_id(post_id)
title = self.validate_text(
title,
"title"
)
return self.client.patch(
f"/posts/{post_id}",
{
"title": title
}
)
def delete_post(self, post_id):
"""
Delete one post.
"""
self.validate_id(post_id)
return self.client.delete(
f"/posts/{post_id}"
)
@staticmethod
def validate_id(value):
"""
Require a positive integer identifier.
"""
if not isinstance(value, int):
raise ValueError(
"ID must be an integer."
)
if value <= 0:
raise ValueError(
"ID must be positive."
)
@staticmethod
def validate_text(value, field_name):
"""
Require non-empty text.
"""
if not isinstance(value, str):
raise ValueError(
f"{field_name} must be text."
)
value = value.strip()
if not value:
raise ValueError(
f"{field_name} cannot be empty."
)
return value
Step 5: Create app.py
# app.py
# Import the API client and exceptions.
from api_client import (
APIClient,
APIClientError,
APIConnectionError,
APITimeoutError,
APIResponseError
)
# Import configuration.
from config import (
Settings,
validate_settings
)
# Import the post service.
from post_service import PostService
def show_menu():
"""
Display the application menu.
"""
print("\nREST API Post Manager")
print("1. List posts")
print("2. View one post")
print("3. Find posts by user")
print("4. Create post")
print("5. Replace post")
print("6. Update post title")
print("7. Delete post")
print("8. Exit")
def display_post(post):
"""
Display one post clearly.
"""
print("\nPost")
print("ID:", post.get("id"))
print("User ID:", post.get("userId"))
print("Title:", post.get("title"))
print("Body:", post.get("body"))
def display_posts(posts):
"""
Display a short list of posts.
"""
if not posts:
print("No posts were found.")
return
for post in posts:
print(
f'{post.get("id")}: '
f'{post.get("title")}'
)
def read_positive_integer(prompt):
"""
Read and validate a positive integer.
"""
value = int(
input(prompt)
)
if value <= 0:
raise ValueError(
"The number must be positive."
)
return value
def main():
# Validate environment configuration.
validate_settings()
# Create the reusable API client.
client = APIClient(
base_url=Settings.API_BASE_URL,
token=Settings.API_TOKEN,
timeout=(
Settings.CONNECT_TIMEOUT,
Settings.READ_TIMEOUT
)
)
# Create the post service.
service = PostService(client)
try:
while True:
show_menu()
choice = input(
"Choose an option: "
).strip()
try:
if choice == "1":
page = read_positive_integer(
"Page number: "
)
posts = service.list_posts(
page=page,
limit=Settings.PAGE_SIZE
)
print(
f"\nPage {page}"
)
display_posts(posts)
elif choice == "2":
post_id = read_positive_integer(
"Post ID: "
)
post = service.get_post(
post_id
)
display_post(post)
elif choice == "3":
user_id = read_positive_integer(
"User ID: "
)
posts = (
service.get_posts_by_user(
user_id
)
)
display_posts(posts)
elif choice == "4":
title = input(
"Title: "
)
body = input(
"Body: "
)
user_id = read_positive_integer(
"User ID: "
)
post = service.create_post(
title,
body,
user_id
)
print(
"Post created successfully."
)
display_post(post)
elif choice == "5":
post_id = read_positive_integer(
"Post ID: "
)
title = input(
"New title: "
)
body = input(
"New body: "
)
user_id = read_positive_integer(
"User ID: "
)
post = service.replace_post(
post_id,
title,
body,
user_id
)
print(
"Post replaced successfully."
)
display_post(post)
elif choice == "6":
post_id = read_positive_integer(
"Post ID: "
)
title = input(
"New title: "
)
post = service.update_title(
post_id,
title
)
print(
"Post updated successfully."
)
display_post(post)
elif choice == "7":
post_id = read_positive_integer(
"Post ID: "
)
service.delete_post(
post_id
)
print(
"Post deleted successfully."
)
elif choice == "8":
print("Goodbye.")
break
else:
print("Invalid option.")
except ValueError as error:
print("Input error:", error)
except APITimeoutError as error:
print("Timeout error:", error)
except APIConnectionError as error:
print("Connection error:", error)
except APIResponseError as error:
print(
f"API error "
f"{error.status_code}: "
f"{error}"
)
except APIClientError as error:
print("Request error:", error)
finally:
# Always close the HTTP session.
client.close()
if __name__ == "__main__":
main()
Step 6: Optional environment variables
API_BASE_URL=https://jsonplaceholder.typicode.com
API_TOKEN=
CONNECT_TIMEOUT=3
READ_TIMEOUT=10
PAGE_SIZE=10
Step 7: Run the project
python app.py
Example Output:
REST API Post Manager
1. List posts
2. View one post
3. Find posts by user
4. Create post
5. Replace post
6. Update post title
7. Delete post
8. Exit
Choose an option: 1
Page number: 1
Page 1
1: sunt aut facere repellat provident occaecati excepturi optio reprehenderit
2: qui est esse
3: ea molestias quasi exercitationem repellat qui ipsa sit aut
4: eum et est occaecati
5: nesciunt quas odio
6: dolorem eum magni eos aperiam quia
7: magnam facilis autem
8: dolorem dolore est ipsam
9: nesciunt iure omnis dolorem tempora et accusantium
10: optio molestias id quia eum
Choose an option: 2
Post ID: 1
Post
ID: 1
User ID: 1
Title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit
Body: quia et suscipit...
Choose an option: 4
Title: Learning Python APIs
Body: This post was created with requests.
User ID: 1
Post created successfully.
Post
ID: 101
User ID: 1
Title: Learning Python APIs
Body: This post was created with requests.
Choose an option: 6
Post ID: 1
New title: Updated Python API Lesson
Post updated successfully.
Post
ID: 1
User ID: 1
Title: Updated Python API Lesson
Body: quia et suscipit...
Choose an option: 7
Post ID: 1
Post deleted successfully.
Choose an option: 8
Goodbye.
Output Explanation:
The application communicates with a REST API through a reusable client. It sends GET, POST, PUT, PATCH, and DELETE requests. The API used for practice simulates changes and does not permanently store created, updated, or deleted posts.
Project Summary
This project demonstrates REST endpoints, HTTP methods, path parameters, query parameters, request headers, JSON bodies, JSON responses, sessions, bearer tokens, connection and read timeouts, retries, status-code validation, custom exceptions, pagination, input validation, reusable service classes, and clean session shutdown.
