Python Examples

This page provides comprehensive Python examples for interacting with the JSONPlaceholder API using the requests library.

Setup

First, install the required library:

bash

pip install requests

Basic Setup

Choose the appropriate base URL for your use case:

Public APILocal Development

python

import requests
import json
from typing import Dict, List, Optional

BASE_URL = "https://api.jsonplaceholder.dev"

class JSONPlaceholderAPI:
    def __init__(self, base_url: str = BASE_URL):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'User-Agent': 'JSONPlaceholder-Python-Client/1.0'
        })

    def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response

    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
        response = self._make_request('GET', endpoint, params=params)
        return response.json()

    def post(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('POST', endpoint, json=data)
        return response.json()

    def put(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('PUT', endpoint, json=data)
        return response.json()

    def patch(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('PATCH', endpoint, json=data)
        return response.json()

    def delete(self, endpoint: str) -> Dict:
        response = self._make_request('DELETE', endpoint)
        return response.json()

# Initialize API client
api = JSONPlaceholderAPI()

python

import requests
import json
from typing import Dict, List, Optional

BASE_URL = "http://localhost:3000"

class JSONPlaceholderAPI:
    def __init__(self, base_url: str = BASE_URL):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'User-Agent': 'JSONPlaceholder-Python-Client/1.0'
        })

    def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response

    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
        response = self._make_request('GET', endpoint, params=params)
        return response.json()

    def post(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('POST', endpoint, json=data)
        return response.json()

    def put(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('PUT', endpoint, json=data)
        return response.json()

    def patch(self, endpoint: str, data: Dict) -> Dict:
        response = self._make_request('PATCH', endpoint, json=data)
        return response.json()

    def delete(self, endpoint: str) -> Dict:
        response = self._make_request('DELETE', endpoint)
        return response.json()

# Initialize API client
api = JSONPlaceholderAPI()

Users Examples

Basic User Operations

python

# Get all users
def get_all_users():
    try:
        users = api.get('/users')
        print(f"Found {len(users)} users")
        return users
    except requests.RequestException as e:
        print(f"Error fetching users: {e}")
        return []

# Get single user
def get_user_by_id(user_id: int):
    try:
        user = api.get(f'/users/{user_id}')
        print(f"User: {user['name']} ({user['email']})")
        return user
    except requests.RequestException as e:
        print(f"Error fetching user {user_id}: {e}")
        return None

# Create new user
def create_user(name: str, username: str, email: str, **kwargs):
    user_data = {
        'name': name,
        'username': username,
        'email': email,
        'address': kwargs.get('address', {
            'street': '123 Main St',
            'city': 'New York',
            'zipcode': '10001'
        }),
        'phone': kwargs.get('phone', '1-555-123-4567'),
        'website': kwargs.get('website', 'example.com')
    }
    
    try:
        new_user = api.post('/users', user_data)
        print(f"Created user: {new_user['name']} (ID: {new_user['id']})")
        return new_user
    except requests.RequestException as e:
        print(f"Error creating user: {e}")
        return None

# Usage examples
users = get_all_users()
user = get_user_by_id(1)
new_user = create_user("John Doe", "johndoe", "[email protected]")

User Management Functions

python

def update_user(user_id: int, **updates):
    """Update user with partial data"""
    try:
        updated_user = api.patch(f'/users/{user_id}', updates)
        print(f"Updated user {user_id}")
        return updated_user
    except requests.RequestException as e:
        print(f"Error updating user {user_id}: {e}")
        return None

def delete_user(user_id: int):
    """Delete a user"""
    try:
        result = api.delete(f'/users/{user_id}')
        print(f"Deleted user {user_id}")
        return result
    except requests.RequestException as e:
        print(f"Error deleting user {user_id}: {e}")
        return None

def get_user_posts(user_id: int):
    """Get all posts by a user"""
    try:
        posts = api.get(f'/users/{user_id}/posts')
        print(f"User {user_id} has {len(posts)} posts")
        return posts
    except requests.RequestException as e:
        print(f"Error fetching posts for user {user_id}: {e}")
        return []

def get_user_stats(user_id: int):
    """Get comprehensive user statistics"""
    try:
        user = api.get(f'/users/{user_id}')
        posts = api.get(f'/users/{user_id}/posts')
        albums = api.get(f'/users/{user_id}/albums')
        todos = api.get(f'/users/{user_id}/todos')
        
        stats = {
            'user': user,
            'posts_count': len(posts),
            'albums_count': len(albums),
            'todos_count': len(todos),
            'completed_todos': len([t for t in todos if t['completed']]),
            'completion_rate': len([t for t in todos if t['completed']]) / len(todos) * 100 if todos else 0
        }
        
        print(f"Stats for {user['name']}:")
        print(f"  Posts: {stats['posts_count']}")
        print(f"  Albums: {stats['albums_count']}")
        print(f"  Todos: {stats['todos_count']} ({stats['completion_rate']:.1f}% complete)")
        
        return stats
    except requests.RequestException as e:
        print(f"Error fetching stats for user {user_id}: {e}")
        return None

# Usage
update_user(1, email="[email protected]")
user_stats = get_user_stats(1)

Posts Examples

Post Management

python

def get_all_posts():
    """Get all posts"""
    return api.get('/posts')

def get_posts_by_user(user_id: int):
    """Get posts by specific user"""
    return api.get('/posts', params={'userId': user_id})

def get_post_by_id(post_id: int):
    """Get single post"""
    return api.get(f'/posts/{post_id}')

def create_post(user_id: int, title: str, body: str):
    """Create new post"""
    post_data = {
        'userId': user_id,
        'title': title,
        'body': body
    }
    return api.post('/posts', post_data)

def update_post(post_id: int, **updates):
    """Update post with partial data"""
    return api.patch(f'/posts/{post_id}', updates)

def delete_post(post_id: int):
    """Delete post"""
    return api.delete(f'/posts/{post_id}')

def get_post_comments(post_id: int):
    """Get all comments for a post"""
    return api.get(f'/posts/{post_id}/comments')

# Usage examples
posts = get_all_posts()
user_posts = get_posts_by_user(1)
new_post = create_post(1, "My Python Post", "This post was created using Python!")
post_comments = get_post_comments(1)

Advanced Post Operations

python

def search_posts(search_term: str):
    """Search posts by title or content"""
    posts = get_all_posts()
    matching_posts = [
        post for post in posts
        if search_term.lower() in post['title'].lower() 
        or search_term.lower() in post['body'].lower()
    ]
    return matching_posts

def get_popular_posts(min_comments: int = 3):
    """Get posts with at least N comments"""
    posts = get_all_posts()
    popular_posts = []
    
    for post in posts:
        comments = get_post_comments(post['id'])
        if len(comments) >= min_comments:
            post['comment_count'] = len(comments)
            popular_posts.append(post)
    
    return sorted(popular_posts, key=lambda x: x['comment_count'], reverse=True)

def bulk_create_posts(posts_data: List[Dict]):
    """Create multiple posts"""
    created_posts = []
    for post_data in posts_data:
        try:
            created_post = create_post(**post_data)
            created_posts.append(created_post)
            print(f"Created post: {created_post['title']}")
        except Exception as e:
            print(f"Failed to create post '{post_data.get('title', 'Unknown')}': {e}")
    
    return created_posts

# Usage
search_results = search_posts("python")
popular = get_popular_posts(4)

bulk_posts = [
    {'user_id': 1, 'title': 'Post 1', 'body': 'Content 1'},
    {'user_id': 1, 'title': 'Post 2', 'body': 'Content 2'},
    {'user_id': 2, 'title': 'Post 3', 'body': 'Content 3'},
]
created_posts = bulk_create_posts(bulk_posts)

Comments Examples

python

def get_all_comments():
    """Get all comments"""
    return api.get('/comments')

def get_comments_by_post(post_id: int):
    """Get comments for specific post"""
    return api.get('/comments', params={'postId': post_id})

def create_comment(post_id: int, name: str, email: str, body: str):
    """Create new comment"""
    comment_data = {
        'postId': post_id,
        'name': name,
        'email': email,
        'body': body
    }
    return api.post('/comments', comment_data)

def update_comment(comment_id: int, **updates):
    """Update comment"""
    return api.patch(f'/comments/{comment_id}', updates)

def delete_comment(comment_id: int):
    """Delete comment"""
    return api.delete(f'/comments/{comment_id}')

def moderate_comments(keywords: List[str]):
    """Find comments containing specific keywords (content moderation)"""
    comments = get_all_comments()
    flagged_comments = []
    
    for comment in comments:
        for keyword in keywords:
            if keyword.lower() in comment['body'].lower():
                flagged_comments.append({
                    'comment': comment,
                    'flagged_for': keyword
                })
                break
    
    return flagged_comments

# Usage
comments = get_all_comments()
post_comments = get_comments_by_post(1)
new_comment = create_comment(1, "Great post!", "[email protected]", "Thanks for sharing!")
flagged = moderate_comments(["spam", "inappropriate"])

Advanced Examples

Data Analysis

python

import statistics
from collections import defaultdict

def analyze_user_activity():
    """Analyze user activity across the platform"""
    users = get_all_users()
    posts = get_all_posts()
    comments = get_all_comments()
    todos = api.get('/todos')
    
    # Group data by user
    user_activity = defaultdict(lambda: {
        'posts': 0, 'comments': 0, 'todos': 0, 'completed_todos': 0
    })
    
    # Count posts per user
    for post in posts:
        user_activity[post['userId']]['posts'] += 1
    
    # Count todos per user
    for todo in todos:
        user_activity[todo['userId']]['todos'] += 1
        if todo['completed']:
            user_activity[todo['userId']]['completed_todos'] += 1
    
    # Calculate statistics
    post_counts = [activity['posts'] for activity in user_activity.values()]
    todo_counts = [activity['todos'] for activity in user_activity.values()]
    
    analysis = {
        'total_users': len(users),
        'total_posts': len(posts),
        'total_comments': len(comments),
        'total_todos': len(todos),
        'avg_posts_per_user': statistics.mean(post_counts) if post_counts else 0,
        'avg_todos_per_user': statistics.mean(todo_counts) if todo_counts else 0,
        'most_active_users': sorted(
            user_activity.items(),
            key=lambda x: x[1]['posts'],
            reverse=True
        )[:5]
    }
    
    print("Platform Analysis:")
    print(f"Total Users: {analysis['total_users']}")
    print(f"Total Posts: {analysis['total_posts']}")
    print(f"Total Comments: {analysis['total_comments']}")
    print(f"Average Posts per User: {analysis['avg_posts_per_user']:.1f}")
    print(f"Average Todos per User: {analysis['avg_todos_per_user']:.1f}")
    
    return analysis

def export_user_data_to_csv(user_id: int, filename: str):
    """Export all user data to CSV"""
    import csv
    
    user = get_user_by_id(user_id)
    posts = get_user_posts(user_id)
    albums = api.get(f'/users/{user_id}/albums')
    todos = api.get(f'/users/{user_id}/todos')
    
    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.writer(csvfile)
        
        # User info
        writer.writerow(['User Information'])
        writer.writerow(['Name', user['name']])
        writer.writerow(['Email', user['email']])
        writer.writerow(['Phone', user['phone']])
        writer.writerow([])
        
        # Posts
        writer.writerow(['Posts'])
        writer.writerow(['ID', 'Title', 'Body'])
        for post in posts:
            writer.writerow([post['id'], post['title'], post['body']])
        writer.writerow([])
        
        # Todos
        writer.writerow(['Todos'])
        writer.writerow(['ID', 'Title', 'Completed'])
        for todo in todos:
            writer.writerow([todo['id'], todo['title'], todo['completed']])
    
    print(f"User data exported to {filename}")

# Usage
analysis = analyze_user_activity()
export_user_data_to_csv(1, "user_1_data.csv")

Async Operations (using aiohttp)

python

import asyncio
import aiohttp
from typing import List

class AsyncJSONPlaceholderAPI:
    def __init__(self, base_url: str = BASE_URL):
        self.base_url = base_url

    async def _make_request(self, session: aiohttp.ClientSession, method: str, endpoint: str, **kwargs):
        url = f"{self.base_url}{endpoint}"
        async with session.request(method, url, **kwargs) as response:
            response.raise_for_status()
            return await response.json()

    async def get_multiple_users(self, user_ids: List[int]):
        """Fetch multiple users concurrently"""
        async with aiohttp.ClientSession() as session:
            tasks = [
                self._make_request(session, 'GET', f'/users/{user_id}')
                for user_id in user_ids
            ]
            return await asyncio.gather(*tasks)

    async def get_user_with_posts(self, user_id: int):
        """Fetch user and their posts concurrently"""
        async with aiohttp.ClientSession() as session:
            user_task = self._make_request(session, 'GET', f'/users/{user_id}')
            posts_task = self._make_request(session, 'GET', f'/users/{user_id}/posts')
            albums_task = self._make_request(session, 'GET', f'/users/{user_id}/albums')
            todos_task = self._make_request(session, 'GET', f'/users/{user_id}/todos')
            
            user, posts, albums, todos = await asyncio.gather(
                user_task, posts_task, albums_task, todos_task
            )
            
            return {
                'user': user,
                'posts': posts,
                'albums': albums,
                'todos': todos
            }

# Usage
async def main():
    async_api = AsyncJSONPlaceholderAPI()
    
    # Fetch multiple users concurrently
    users = await async_api.get_multiple_users([1, 2, 3, 4, 5])
    print(f"Fetched {len(users)} users concurrently")
    
    # Fetch user with all related data
    user_data = await async_api.get_user_with_posts(1)
    print(f"User: {user_data['user']['name']}")
    print(f"Posts: {len(user_data['posts'])}")
    print(f"Albums: {len(user_data['albums'])}")
    print(f"Todos: {len(user_data['todos'])}")

# Run async example
# asyncio.run(main())

Error Handling and Retry Logic

python

import time
from functools import wraps

def retry(max_attempts=3, delay=1, backoff=2):
    """Decorator for retrying failed requests"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            current_delay = delay
            
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except requests.RequestException as e:
                    attempts += 1
                    if attempts == max_attempts:
                        print(f"Failed after {max_attempts} attempts: {e}")
                        raise
                    
                    print(f"Attempt {attempts} failed: {e}. Retrying in {current_delay}s...")
                    time.sleep(current_delay)
                    current_delay *= backoff
            
        return wrapper
    return decorator

class RobustAPIClient(JSONPlaceholderAPI):
    @retry(max_attempts=3)
    def robust_get(self, endpoint: str, **kwargs):
        """GET request with retry logic"""
        return super().get(endpoint, **kwargs)
    
    @retry(max_attempts=3)
    def robust_post(self, endpoint: str, data: Dict):
        """POST request with retry logic"""
        return super().post(endpoint, data)

# Usage
robust_api = RobustAPIClient()
users = robust_api.robust_get('/users')

Performance Testing

python

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def benchmark_api_performance():
    """Benchmark API performance with different concurrency levels"""
    def make_request(endpoint):
        start_time = time.time()
        try:
            response = api.get(endpoint)
            return time.time() - start_time, len(response)
        except Exception as e:
            return time.time() - start_time, 0

    endpoints = [f'/users/{i}' for i in range(1, 11)]
    
    # Sequential requests
    start_time = time.time()
    sequential_times = []
    for endpoint in endpoints:
        duration, _ = make_request(endpoint)
        sequential_times.append(duration)
    sequential_total = time.time() - start_time
    
    # Concurrent requests
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=5) as executor:
        future_to_endpoint = {executor.submit(make_request, endpoint): endpoint 
                             for endpoint in endpoints}
        concurrent_times = []
        for future in as_completed(future_to_endpoint):
            duration, _ = future.result()
            concurrent_times.append(duration)
    concurrent_total = time.time() - start_time
    
    print("Performance Benchmark Results:")
    print(f"Sequential: {sequential_total:.2f}s total, {statistics.mean(sequential_times):.3f}s avg")
    print(f"Concurrent: {concurrent_total:.2f}s total, {statistics.mean(concurrent_times):.3f}s avg")
    print(f"Speed improvement: {sequential_total/concurrent_total:.2f}x faster")

# Run benchmark
benchmark_api_performance()

Best Practices

Always handle exceptions - Network requests can fail
Use session objects - For connection pooling and performance
Implement retry logic - For resilient applications
Validate data - Check response structure before using
Use type hints - For better code documentation
Cache responses - When appropriate to reduce API calls

Python Examples ​

Setup ​

Basic Setup ​

Users Examples ​

Basic User Operations ​

User Management Functions ​

Posts Examples ​

Post Management ​

Advanced Post Operations ​

Comments Examples ​

Advanced Examples ​

Data Analysis ​

Async Operations (using aiohttp) ​

Error Handling and Retry Logic ​

Performance Testing ​

Best Practices ​

Python Examples

Setup

Basic Setup

Users Examples

Basic User Operations

User Management Functions

Posts Examples

Post Management

Advanced Post Operations

Comments Examples

Advanced Examples

Data Analysis

Async Operations (using aiohttp)

Error Handling and Retry Logic

Performance Testing

Best Practices