In this comprehensive tutorial, we’ll learn how to consume a REST API using Python entirely from the command line. We’ll build a complete project covering all CRUD operations using the free JSONPlaceholder API.
Project Overview
Folder Structure
python-api-client/ │ ├── api_client.py # HTTP client and API operations ├── models.py # Data models ├── main.py # Main application with CLI interface ├── utils.py # Utility functions ├── requirements.txt # Dependencies ├── README.md # Documentation └── run.sh # Run script (optional)
Step 1: Project Setup
First, create the project structure:
# Create main project directory mkdir python-api-client cd python-api-client # Create all Python files touch api_client.py models.py main.py utils.py requirements.txt README.md run.sh chmod +x run.sh # Create a virtual environment python3 -m venv venv # Activate virtual environment # On Linux/Mac: source venv/bin/activate # On Windows: # venv\Scripts\activate
Step 2: Install Dependencies
Edit requirements.txt:
requests==2.31.0 tabulate==0.9.0 colorama==0.4.6
Install the dependencies:
pip install -r requirements.txt
Step 3: Create Data Models (models.py)
Create models.py:
"""
Data models for the API client application.
"""
from dataclasses import dataclass, asdict
from typing import Optional, List
import json
@dataclass
class Post:
"""Post data model representing a blog post."""
user_id: int
id: Optional[int] = None
title: Optional[str] = None
body: Optional[str] = None
def to_dict(self) -> dict:
"""Convert Post object to dictionary, excluding None values."""
return {k: v for k, v in asdict(self).items() if v is not None}
def to_json(self, indent: int = 2) -> str:
"""Convert Post object to JSON string."""
return json.dumps(self.to_dict(), indent=indent)
@classmethod
def from_dict(cls, data: dict) -> 'Post':
"""Create Post object from dictionary."""
# Handle both camelCase (API) and snake_case (Python) keys
user_id = data.get('userId', data.get('user_id'))
return cls(
user_id=user_id if user_id is not None else 1,
id=data.get('id'),
title=data.get('title'),
body=data.get('body')
)
@classmethod
def from_json(cls, json_str: str) -> 'Post':
"""Create Post object from JSON string."""
data = json.loads(json_str)
return cls.from_dict(data)
def display(self, detailed: bool = False) -> str:
"""Return formatted string representation of the post."""
if detailed:
return f"""
POST DETAILS
============
ID: {self.id if self.id else 'NEW'}
User ID: {self.user_id}
Title: {self.title if self.title else 'N/A'}
Body: {self.body if self.body else 'N/A'}
"""
else:
title_preview = (self.title or 'N/A')[:50]
body_preview = (self.body or 'N/A')[:100]
return f"ID: {self.id}, User: {self.user_id}, Title: {title_preview}..."
@dataclass
class User:
"""User data model."""
id: int
name: str
username: str
email: str
phone: Optional[str] = None
website: Optional[str] = None
address: Optional[dict] = None
company: Optional[dict] = None
def display(self, detailed: bool = False) -> str:
"""Return formatted string representation of the user."""
if detailed:
address_str = json.dumps(self.address, indent=2) if self.address else 'N/A'
company_str = json.dumps(self.company, indent=2) if self.company else 'N/A'
return f"""
USER DETAILS
============
ID: {self.id}
Name: {self.name}
Username: {self.username}
Email: {self.email}
Phone: {self.phone or 'N/A'}
Website: {self.website or 'N/A'}
Address: {address_str}
Company: {company_str}
"""
else:
return f"ID: {self.id}, Name: {self.name}, Email: {self.email}"
@classmethod
def from_dict(cls, data: dict) -> 'User':
"""Create User object from dictionary."""
return cls(
id=data.get('id'),
name=data.get('name', ''),
username=data.get('username', ''),
email=data.get('email', ''),
phone=data.get('phone'),
website=data.get('website'),
address=data.get('address'),
company=data.get('company')
)
Step 4: Create API Client (api_client.py)
Create api_client.py:
"""
API client for interacting with JSONPlaceholder API.
Handles all HTTP requests and error handling.
"""
import requests
import json
from typing import Optional, List, Dict, Any
from models import Post, User
class JSONPlaceholderClient:
"""Client for interacting with JSONPlaceholder API."""
BASE_URL = "https://jsonplaceholder.typicode.com"
TIMEOUT = 10 # seconds
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'Python-API-Client/1.0'
})
def _log_request(self, method: str, endpoint: str, status_code: Optional[int] = None):
"""Log HTTP request details."""
if status_code:
print(f"[{method}] {endpoint} -> Status: {status_code}")
else:
print(f"[{method}] {endpoint}")
def _handle_response(self, response: requests.Response) -> Dict[str, Any]:
"""Handle API response and return JSON data."""
try:
response.raise_for_status()
return response.json() if response.content else {}
except requests.exceptions.HTTPError as http_err:
print(f"HTTP Error: {http_err}")
print(f"Response: {response.text}")
return {}
except json.JSONDecodeError as json_err:
print(f"JSON Decode Error: {json_err}")
return {}
except Exception as err:
print(f"Request Error: {err}")
return {}
def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request to API."""
url = f"{self.BASE_URL}{endpoint}"
# Set default timeout
kwargs.setdefault('timeout', self.TIMEOUT)
try:
response = self.session.request(method, url, **kwargs)
self._log_request(method, endpoint, response.status_code)
return self._handle_response(response)
except requests.exceptions.Timeout:
print(f"Timeout Error: Request to {endpoint} timed out after {self.TIMEOUT} seconds")
return {}
except requests.exceptions.ConnectionError:
print(f"Connection Error: Unable to connect to {url}")
return {}
except Exception as e:
print(f"Unexpected Error: {e}")
return {}
# CRUD Operations for Posts
def get_all_posts(self) -> List[Post]:
"""Get all posts."""
data = self._make_request('GET', '/posts')
if isinstance(data, list):
return [Post.from_dict(item) for item in data]
return []
def get_post(self, post_id: int) -> Optional[Post]:
"""Get a specific post by ID."""
data = self._make_request('GET', f'/posts/{post_id}')
return Post.from_dict(data) if data else None
def create_post(self, post: Post) -> Optional[Post]:
"""Create a new post."""
data = self._make_request('POST', '/posts', json=post.to_dict())
return Post.from_dict(data) if data else None
def update_post(self, post_id: int, post: Post) -> Optional[Post]:
"""Update an entire post (PUT)."""
data = self._make_request('PUT', f'/posts/{post_id}', json=post.to_dict())
return Post.from_dict(data) if data else None
def patch_post(self, post_id: int, updates: Dict[str, Any]) -> Optional[Post]:
"""Partially update a post (PATCH)."""
data = self._make_request('PATCH', f'/posts/{post_id}', json=updates)
return Post.from_dict(data) if data else None
def delete_post(self, post_id: int) -> bool:
"""Delete a post."""
data = self._make_request('DELETE', f'/posts/{post_id}')
# DELETE typically returns empty object on success
return data is not None
# CRUD Operations for Users
def get_all_users(self) -> List[User]:
"""Get all users."""
data = self._make_request('GET', '/users')
if isinstance(data, list):
return [User.from_dict(item) for item in data]
return []
def get_user(self, user_id: int) -> Optional[User]:
"""Get a specific user by ID."""
data = self._make_request('GET', f'/users/{user_id}')
return User.from_dict(data) if data else None
def get_user_posts(self, user_id: int) -> List[Post]:
"""Get all posts by a specific user."""
data = self._make_request('GET', f'/posts?userId={user_id}')
if isinstance(data, list):
return [Post.from_dict(item) for item in data]
return []
# Search operations
def search_posts(self, title: str = None, body: str = None) -> List[Post]:
"""Search posts by title or body content."""
all_posts = self.get_all_posts()
if not all_posts:
return []
results = []
for post in all_posts:
match = True
if title and title.lower() not in (post.title or '').lower():
match = False
if body and body.lower() not in (post.body or '').lower():
match = False
if match:
results.append(post)
return results
# Resource validation
def validate_post_id(self, post_id: int) -> bool:
"""Check if a post ID exists."""
post = self.get_post(post_id)
return post is not None and post.id is not None
def validate_user_id(self, user_id: int) -> bool:
"""Check if a user ID exists."""
user = self.get_user(user_id)
return user is not None and user.id is not None
# Create a singleton instance
api_client = JSONPlaceholderClient()
Step 5: Create Utility Functions (utils.py)
Create utils.py:
"""
Utility functions for the API client application.
Includes display functions, input handlers, and formatting utilities.
"""
import sys
import json
from typing import List, Optional, Any
from tabulate import tabulate
from colorama import init, Fore, Style, Back
# Initialize colorama for cross-platform colored output
init(autoreset=True)
class DisplayUtils:
"""Utility class for displaying data in the console."""
@staticmethod
def print_success(message: str):
"""Print success message in green."""
print(f"{Fore.GREEN}[SUCCESS]{Style.RESET_ALL} {message}")
@staticmethod
def print_error(message: str):
"""Print error message in red."""
print(f"{Fore.RED}[ERROR]{Style.RESET_ALL} {message}")
@staticmethod
def print_info(message: str):
"""Print info message in blue."""
print(f"{Fore.BLUE}[INFO]{Style.RESET_ALL} {message}")
@staticmethod
def print_warning(message: str):
"""Print warning message in yellow."""
print(f"{Fore.YELLOW}[WARNING]{Style.RESET_ALL} {message}")
@staticmethod
def print_header(text: str, width: int = 60):
"""Print a formatted header."""
print(f"\n{Fore.CYAN}{'=' * width}")
print(f"{text.center(width)}")
print(f"{'=' * width}{Style.RESET_ALL}\n")
@staticmethod
def print_subheader(text: str):
"""Print a subheader."""
print(f"\n{Fore.MAGENTA}{text}")
print(f"{'-' * len(text)}{Style.RESET_ALL}")
@staticmethod
def display_posts_table(posts: List, title: str = "Posts", limit: int = 10):
"""Display posts in a formatted table."""
if not posts:
DisplayUtils.print_info("No posts to display.")
return
DisplayUtils.print_header(title)
# Prepare table data
table_data = []
for i, post in enumerate(posts[:limit]):
title_preview = (post.title or 'N/A')[:40] + '...' if len(post.title or '') > 40 else post.title or 'N/A'
body_preview = (post.body or 'N/A')[:60] + '...' if len(post.body or '') > 60 else post.body or 'N/A'
table_data.append([
post.id or 'NEW',
post.user_id,
title_preview,
body_preview
])
headers = ["ID", "User ID", "Title", "Body Preview"]
print(tabulate(table_data, headers=headers, tablefmt="grid"))
if len(posts) > limit:
print(f"\n{Fore.YELLOW}Showing {limit} of {len(posts)} posts. Use specific ID to view more.{Style.RESET_ALL}")
@staticmethod
def display_users_table(users: List):
"""Display users in a formatted table."""
if not users:
DisplayUtils.print_info("No users to display.")
return
DisplayUtils.print_header("Users")
table_data = []
for user in users:
table_data.append([
user.id,
user.name[:20],
user.username,
user.email[:20] + '...' if len(user.email) > 20 else user.email,
user.phone or 'N/A'
])
headers = ["ID", "Name", "Username", "Email", "Phone"]
print(tabulate(table_data, headers=headers, tablefmt="grid"))
@staticmethod
def display_post_detail(post):
"""Display detailed view of a single post."""
DisplayUtils.print_header(f"Post Details - ID: {post.id}")
data = [
["Field", "Value"],
["ID", post.id or "NEW"],
["User ID", post.user_id],
["Title", post.title or "N/A"],
["Body", post.body or "N/A"]
]
print(tabulate(data, headers="firstrow", tablefmt="grid"))
@staticmethod
def display_json(data: Any, title: str = "JSON Response"):
"""Display JSON data in formatted way."""
DisplayUtils.print_subheader(title)
if isinstance(data, (dict, list)):
print(json.dumps(data, indent=2))
else:
print(data)
class InputHandler:
"""Handles user input with validation."""
@staticmethod
def get_int_input(prompt: str, min_val: Optional[int] = None,
max_val: Optional[int] = None) -> int:
"""Get integer input with validation."""
while True:
try:
value = input(f"{Fore.CYAN}{prompt}: {Style.RESET_ALL}").strip()
if not value:
DisplayUtils.print_error("Input cannot be empty.")
continue
value_int = int(value)
if min_val is not None and value_int < min_val:
DisplayUtils.print_error(f"Value must be at least {min_val}")
continue
if max_val is not None and value_int > max_val:
DisplayUtils.print_error(f"Value must be at most {max_val}")
continue
return value_int
except ValueError:
DisplayUtils.print_error("Please enter a valid integer.")
@staticmethod
def get_string_input(prompt: str, required: bool = True,
min_length: int = 1) -> str:
"""Get string input with validation."""
while True:
value = input(f"{Fore.CYAN}{prompt}: {Style.RESET_ALL}").strip()
if required and not value:
DisplayUtils.print_error("Input cannot be empty.")
continue
if min_length > 1 and len(value) < min_length:
DisplayUtils.print_error(f"Input must be at least {min_length} characters.")
continue
return value
@staticmethod
def get_yes_no_input(prompt: str) -> bool:
"""Get yes/no input."""
while True:
response = input(f"{Fore.CYAN}{prompt} (yes/no): {Style.RESET_ALL}").strip().lower()
if response in ['y', 'yes']:
return True
elif response in ['n', 'no']:
return False
else:
DisplayUtils.print_error("Please enter 'yes' or 'no'.")
@staticmethod
def get_multiline_input(prompt: str) -> str:
"""Get multiline input from user."""
DisplayUtils.print_info(prompt)
DisplayUtils.print_info("Enter your text (press Enter twice to finish):")
lines = []
while True:
try:
line = input()
if line == "" and lines and lines[-1] == "":
break
lines.append(line)
except EOFError:
break
# Remove the last empty line
if lines and lines[-1] == "":
lines.pop()
return "\n".join(lines)
class ValidationUtils:
"""Utility class for data validation."""
@staticmethod
def validate_post_data(title: str, body: str, user_id: int) -> tuple[bool, list]:
"""Validate post data before sending to API."""
errors = []
if not title or len(title.strip()) < 3:
errors.append("Title must be at least 3 characters long")
if not body or len(body.strip()) < 10:
errors.append("Body must be at least 10 characters long")
if user_id <= 0:
errors.append("User ID must be positive")
return len(errors) == 0, errors
@staticmethod
def validate_user_id(user_id: int) -> bool:
"""Validate user ID."""
return isinstance(user_id, int) and user_id > 0
@staticmethod
def validate_post_id(post_id: int) -> bool:
"""Validate post ID."""
return isinstance(post_id, int) and post_id > 0
def clear_screen():
"""Clear the terminal screen."""
print("\033c", end="")
Step 6: Create Main Application (main.py)
Create main.py:
#!/usr/bin/env python3
"""
JSONPlaceholder API Client - Command Line Interface
A complete CRUD application for interacting with JSONPlaceholder API.
"""
import sys
import time
from typing import Optional
from api_client import api_client
from models import Post
from utils import DisplayUtils, InputHandler, ValidationUtils, clear_screen
class ApiConsumerApp:
"""Main application class for the API client."""
def __init__(self):
self.running = True
self.display = DisplayUtils()
self.input_handler = InputHandler()
self.validator = ValidationUtils()
def run(self):
"""Run the main application loop."""
self.display_welcome()
while self.running:
try:
self.main_menu()
except KeyboardInterrupt:
self.display.print_warning("\nInterrupted by user")
if self.input_handler.get_yes_no_input("Are you sure you want to exit?"):
self.running = False
except Exception as e:
self.display.print_error(f"Unexpected error: {e}")
if not self.input_handler.get_yes_no_input("Continue running?"):
self.running = False
self.exit_app()
def display_welcome(self):
"""Display welcome message and application info."""
clear_screen()
self.display.print_header("JSONPlaceholder API Client")
print("A complete CRUD application demonstrating API consumption.")
print("Using free fake API: https://jsonplaceholder.typicode.com")
print("\nThis application supports all CRUD operations:")
print(" - CREATE: Add new posts")
print(" - READ: Retrieve posts and users")
print(" - UPDATE: Modify existing posts")
print(" - DELETE: Remove posts")
print("\n" + "="*60)
time.sleep(2)
def main_menu(self):
"""Display and handle main menu."""
clear_screen()
self.display.print_header("MAIN MENU")
menu_options = [
"1. Get all posts",
"2. Get a specific post by ID",
"3. Create a new post",
"4. Update a post (PUT - full update)",
"5. Partially update a post (PATCH)",
"6. Delete a post",
"7. Get all users",
"8. Get a specific user by ID",
"9. Get all posts by user ID",
"10. Search posts",
"11. Exit application"
]
for option in menu_options:
print(option)
print("\n" + "-"*40)
try:
choice = self.input_handler.get_int_input("Enter your choice (1-11)", 1, 11)
self.handle_menu_choice(choice)
except ValueError as e:
self.display.print_error(f"Invalid input: {e}")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def handle_menu_choice(self, choice: int):
"""Handle menu choice selection."""
operations = {
1: self.get_all_posts,
2: self.get_post_by_id,
3: self.create_post,
4: self.update_post,
5: self.partially_update_post,
6: self.delete_post,
7: self.get_all_users,
8: self.get_user_by_id,
9: self.get_user_posts,
10: self.search_posts,
11: self.exit_app
}
operation = operations.get(choice)
if operation:
operation()
else:
self.display.print_error("Invalid choice selected")
# CRUD Operations Implementation
def get_all_posts(self):
"""Get all posts from the API."""
self.display.print_header("GET ALL POSTS")
self.display.print_info("Fetching all posts from API...")
posts = api_client.get_all_posts()
if posts:
self.display.display_posts_table(posts, "All Posts", limit=15)
self.display.print_info(f"Total posts retrieved: {len(posts)}")
else:
self.display.print_error("Failed to retrieve posts or no posts found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def get_post_by_id(self):
"""Get a specific post by ID."""
self.display.print_header("GET POST BY ID")
post_id = self.input_handler.get_int_input("Enter post ID", min_val=1)
self.display.print_info(f"Fetching post with ID {post_id}...")
post = api_client.get_post(post_id)
if post and post.id:
self.display.display_post_detail(post)
else:
self.display.print_error(f"Post with ID {post_id} not found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def create_post(self):
"""Create a new post."""
self.display.print_header("CREATE NEW POST")
# Get post data from user
user_id = self.input_handler.get_int_input("Enter user ID", min_val=1)
title = self.input_handler.get_string_input("Enter post title", min_length=3)
self.display.print_info("Enter post body (multiline input):")
body = self.input_handler.get_multiline_input("Post Body")
# Validate data
is_valid, errors = self.validator.validate_post_data(title, body, user_id)
if not is_valid:
for error in errors:
self.display.print_error(error)
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Create post object
new_post = Post(user_id=user_id, title=title, body=body)
# Display preview
self.display.print_subheader("Post Preview")
self.display.display_post_detail(new_post)
# Confirm creation
if not self.input_handler.get_yes_no_input("Create this post?"):
self.display.print_info("Post creation cancelled.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Send to API
self.display.print_info("Creating post via API...")
created_post = api_client.create_post(new_post)
if created_post and created_post.id:
self.display.print_success(f"Post created successfully with ID: {created_post.id}")
self.display.display_post_detail(created_post)
else:
self.display.print_error("Failed to create post.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def update_post(self):
"""Update an entire post using PUT."""
self.display.print_header("UPDATE POST (PUT)")
post_id = self.input_handler.get_int_input("Enter post ID to update", min_val=1)
# First, get the existing post
self.display.print_info(f"Fetching existing post with ID {post_id}...")
existing_post = api_client.get_post(post_id)
if not existing_post or not existing_post.id:
self.display.print_error(f"Post with ID {post_id} not found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
self.display.print_subheader("Current Post Data")
self.display.display_post_detail(existing_post)
# Get new data from user
self.display.print_subheader("Enter New Post Data")
user_id = self.input_handler.get_int_input(f"Enter new user ID (current: {existing_post.user_id})", min_val=1)
title = self.input_handler.get_string_input(f"Enter new title (current: {existing_post.title})", min_length=3)
self.display.print_info("Enter new post body:")
body = self.input_handler.get_multiline_input("New Body")
# Validate data
is_valid, errors = self.validator.validate_post_data(title, body, user_id)
if not is_valid:
for error in errors:
self.display.print_error(error)
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Create updated post object
updated_post = Post(user_id=user_id, title=title, body=body)
updated_post.id = post_id
# Display preview
self.display.print_subheader("Updated Post Preview")
self.display.display_post_detail(updated_post)
# Confirm update
if not self.input_handler.get_yes_no_input("Update this post?"):
self.display.print_info("Post update cancelled.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Send PUT request
self.display.print_info(f"Updating post {post_id} via PUT...")
result_post = api_client.update_post(post_id, updated_post)
if result_post:
self.display.print_success(f"Post {post_id} updated successfully.")
self.display.display_post_detail(result_post)
else:
self.display.print_error(f"Failed to update post {post_id}.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def partially_update_post(self):
"""Partially update a post using PATCH."""
self.display.print_header("PARTIALLY UPDATE POST (PATCH)")
post_id = self.input_handler.get_int_input("Enter post ID to update", min_val=1)
# Check if post exists
if not api_client.validate_post_id(post_id):
self.display.print_error(f"Post with ID {post_id} not found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Get existing post for reference
existing_post = api_client.get_post(post_id)
if existing_post:
self.display.print_subheader("Current Post Data")
self.display.display_post_detail(existing_post)
# Ask what fields to update
self.display.print_subheader("Select Fields to Update")
print("1. Update title only")
print("2. Update body only")
print("3. Update both title and body")
choice = self.input_handler.get_int_input("Enter choice (1-3)", 1, 3)
updates = {}
if choice in [1, 3]:
new_title = self.input_handler.get_string_input("Enter new title", min_length=3)
updates['title'] = new_title
if choice in [2, 3]:
self.display.print_info("Enter new post body:")
new_body = self.input_handler.get_multiline_input("New Body")
updates['body'] = new_body
# Validate at least one field is being updated
if not updates:
self.display.print_error("No fields selected for update.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Confirm update
self.display.print_subheader("Updates to Apply")
for key, value in updates.items():
print(f"{key}: {value[:50]}..." if len(str(value)) > 50 else f"{key}: {value}")
if not self.input_handler.get_yes_no_input("Apply these updates?"):
self.display.print_info("Update cancelled.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Send PATCH request
self.display.print_info(f"Partially updating post {post_id} via PATCH...")
result_post = api_client.patch_post(post_id, updates)
if result_post:
self.display.print_success(f"Post {post_id} partially updated successfully.")
self.display.display_post_detail(result_post)
else:
self.display.print_error(f"Failed to update post {post_id}.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def delete_post(self):
"""Delete a post."""
self.display.print_header("DELETE POST")
post_id = self.input_handler.get_int_input("Enter post ID to delete", min_val=1)
# Get post details before deletion
post = api_client.get_post(post_id)
if post:
self.display.print_subheader("Post to be Deleted")
self.display.display_post_detail(post)
else:
self.display.print_warning(f"Post with ID {post_id} may not exist.")
# Confirm deletion
if not self.input_handler.get_yes_no_input(f"Are you sure you want to delete post {post_id}?"):
self.display.print_info("Deletion cancelled.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
# Send DELETE request
self.display.print_info(f"Deleting post {post_id}...")
success = api_client.delete_post(post_id)
if success:
self.display.print_success(f"Post {post_id} deleted successfully.")
self.display.print_info("Note: JSONPlaceholder is a fake API, so the post won't be actually deleted.")
self.display.print_info("But the DELETE operation was successful (returns empty object).")
else:
self.display.print_error(f"Failed to delete post {post_id}.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def get_all_users(self):
"""Get all users from the API."""
self.display.print_header("GET ALL USERS")
self.display.print_info("Fetching all users from API...")
users = api_client.get_all_users()
if users:
self.display.display_users_table(users)
self.display.print_info(f"Total users retrieved: {len(users)}")
else:
self.display.print_error("Failed to retrieve users or no users found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def get_user_by_id(self):
"""Get a specific user by ID."""
self.display.print_header("GET USER BY ID")
user_id = self.input_handler.get_int_input("Enter user ID", min_val=1)
self.display.print_info(f"Fetching user with ID {user_id}...")
user = api_client.get_user(user_id)
if user:
print(user.display(detailed=True))
else:
self.display.print_error(f"User with ID {user_id} not found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def get_user_posts(self):
"""Get all posts by a specific user."""
self.display.print_header("GET USER'S POSTS")
user_id = self.input_handler.get_int_input("Enter user ID", min_val=1)
# Check if user exists
user = api_client.get_user(user_id)
if not user:
self.display.print_error(f"User with ID {user_id} not found.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
self.display.print_info(f"Fetching posts for user: {user.name}...")
posts = api_client.get_user_posts(user_id)
if posts:
self.display.display_posts_table(posts, f"Posts by User {user_id}: {user.name}", limit=15)
self.display.print_info(f"Total posts for user {user_id}: {len(posts)}")
else:
self.display.print_info(f"No posts found for user {user_id}.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def search_posts(self):
"""Search posts by title or body content."""
self.display.print_header("SEARCH POSTS")
print("Search by:")
print("1. Title")
print("2. Body content")
print("3. Both title and body")
choice = self.input_handler.get_int_input("Enter choice (1-3)", 1, 3)
title_search = None
body_search = None
if choice in [1, 3]:
title_search = self.input_handler.get_string_input("Enter title search term", required=False)
if choice in [2, 3]:
body_search = self.input_handler.get_string_input("Enter body search term", required=False)
# Validate at least one search term
if not title_search and not body_search:
self.display.print_error("At least one search term is required.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
return
self.display.print_info("Searching posts...")
results = api_client.search_posts(title=title_search, body=body_search)
if results:
search_terms = []
if title_search:
search_terms.append(f"title: '{title_search}'")
if body_search:
search_terms.append(f"body: '{body_search}'")
self.display.display_posts_table(results, f"Search Results ({', '.join(search_terms)})", limit=15)
self.display.print_info(f"Found {len(results)} matching posts.")
else:
self.display.print_info("No posts found matching your search criteria.")
self.input_handler.get_string_input("Press Enter to continue...", required=False)
def exit_app(self):
"""Exit the application."""
self.display.print_header("EXIT APPLICATION")
if self.input_handler.get_yes_no_input("Are you sure you want to exit?"):
self.display.print_info("Thank you for using the JSONPlaceholder API Client!")
self.display.print_info("Goodbye!")
self.running = False
else:
self.display.print_info("Returning to main menu...")
time.sleep(1)
def main():
"""Main entry point for the application."""
try:
app = ApiConsumerApp()
app.run()
except Exception as e:
print(f"Fatal error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Step 7: Create Run Script and Documentation
Create run.sh:
#!/bin/bash
# Python API Client - Run Script
# Run this script to start the API client application
echo "========================================="
echo " JSONPlaceholder API Client"
echo "========================================="
echo ""
# Check if Python is installed
if ! command -v python3 &> /dev/null; then
echo "ERROR: Python3 is not installed or not in PATH."
echo "Please install Python 3.7 or higher."
exit 1
fi
# Check Python version
PYTHON_VERSION=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')
PYTHON_MAJOR=$(echo $PYTHON_VERSION | cut -d. -f1)
PYTHON_MINOR=$(echo $PYTHON_VERSION | cut -d. -f2)
if [ $PYTHON_MAJOR -lt 3 ] || [ $PYTHON_MAJOR -eq 3 -a $PYTHON_MINOR -lt 7 ]; then
echo "ERROR: Python 3.7 or higher is required."
echo "Current version: $PYTHON_VERSION"
exit 1
fi
# Check if virtual environment exists
if [ ! -d "venv" ]; then
echo "Virtual environment not found. Creating one..."
python3 -m venv venv
echo "Installing dependencies..."
source venv/bin/activate
pip install -r requirements.txt
else
source venv/bin/activate
fi
# Check if requirements are installed
if ! pip show requests &> /dev/null; then
echo "Installing missing dependencies..."
pip install -r requirements.txt
fi
echo ""
echo "Starting API Client..."
echo ""
# Run the application
python3 main.py
# Deactivate virtual environment on exit
deactivate
Create README.md:
# Python API Client for JSONPlaceholder A complete command-line Python application that demonstrates all CRUD operations against the JSONPlaceholder API. ## Features - **Complete CRUD Operations**: - **C**reate: Add new posts - **R**ead: Retrieve posts and users - **U**pdate: Modify existing posts (PUT and PATCH) - **D**elete: Remove posts - **Additional Features**: - User management (list users, get user details) - Search posts by title or content - Get all posts by specific user - Colorful console output with formatted tables - Input validation and error handling - Interactive command-line interface ## Project Structure
python-api-client/
├── api_client.py # HTTP client and API operations
├── models.py # Data models (Post, User)
├── main.py # Main application with CLI interface
├── utils.py # Utility functions for display and input
├── requirements.txt # Python dependencies
├── run.sh # Run script
└── README.md # This file
## Requirements - Python 3.7 or higher - Dependencies listed in `requirements.txt` ## Installation 1. **Clone or create the project structure**:
bash
mkdir python-api-client
cd python-api-client
2. **Set up virtual environment and install dependencies**:
bash
Create virtual environment
python3 -m venv venv
Activate it (Linux/Mac)
source venv/bin/activate
Activate it (Windows)
venv\Scripts\activate
Install dependencies
pip install -r requirements.txt
Or simply use the run script:
bash
chmod +x run.sh
./run.sh
## Usage ### Using the Run Script (Recommended)
bash
./run.sh
### Manual Execution
bash
Activate virtual environment
source venv/bin/activate
Run the application
python3 main.py
### Application Menu When you run the application, you'll see a menu with these options: 1. **Get all posts** - Retrieve and display all posts 2. **Get a specific post by ID** - View details of a single post 3. **Create a new post** - Add a new post to the API 4. **Update a post (PUT)** - Fully replace an existing post 5. **Partially update a post (PATCH)** - Update specific fields only 6. **Delete a post** - Remove a post from the API 7. **Get all users** - Retrieve and display all users 8. **Get a specific user by ID** - View user details 9. **Get all posts by user ID** - View posts from a specific user 10. **Search posts** - Search posts by title or body content 11. **Exit application** - Quit the program ## API Documentation This application uses the [JSONPlaceholder](https://jsonplaceholder.typicode.com/) API: - **Base URL**: `https://jsonplaceholder.typicode.com` - **Posts Endpoint**: `/posts` - **Users Endpoint**: `/users` ### Post Structure
json
{
“userId”: 1,
“id”: 1,
“title”: “Post title”,
“body”: “Post body content”
}
### User Structure
json
{
“id”: 1,
“name”: “User name”,
“username”: “username”,
“email”: “[email protected]”,
“phone”: “1-770-736-8031 x56442”,
“website”: “example.com”
}
## CRUD Operations Explained
### 1. CREATE (POST)
- Endpoint: `POST /posts`
- Creates a new post
- Returns the created post with generated ID
- Example in app: Menu option 3
### 2. READ (GET)
- Get all: `GET /posts`
- Get single: `GET /posts/{id}`
- Retrieves post(s) from the API
- Example in app: Menu options 1 and 2
### 3. UPDATE
- **PUT**: `PUT /posts/{id}` - Replaces entire post
- **PATCH**: `PATCH /posts/{id}` - Updates only specified fields
- Example in app: Menu options 4 and 5
### 4. DELETE
- Endpoint: `DELETE /posts/{id}`
- Removes a post from the API
- Returns empty object on success
- Example in app: Menu option 6
## Error Handling
The application includes comprehensive error handling for:
- Network connectivity issues
- Invalid API responses
- User input validation
- Resource not found errors
- Timeout scenarios
## Testing the Application
Since JSONPlaceholder is a fake API, you can test all operations without affecting real data. The API will return simulated responses.
### Example Workflow
1. **View all posts** (Option 1) to see existing data
2. **Create a new post** (Option 3) with your own data
3. **View your new post** (Option 2) using the returned ID
4. **Update your post** (Option 4 or 5) to modify it
5. **Search for posts** (Option 10) to find specific content
6. **Delete a post** (Option 6) to remove it (simulated)
## Dependencies
- **requests**: HTTP library for API calls
- **tabulate**: Format data as tables in console
- **colorama**: Cross-platform colored terminal text
## License
This is an educational project. Feel free to use and modify as needed.
## Learning Outcomes
By studying this project, you'll learn:
- How to consume REST APIs in Python
- HTTP methods (GET, POST, PUT, PATCH, DELETE)
- JSON serialization/deserialization
- Error handling in API clients
- Building interactive CLI applications
- Structuring Python projects for maintainability
Step 8: Running the Application
Now let’s run our complete application:
# Make the run script executable chmod +x run.sh # Run the application ./run.sh
Or manually:
# Activate virtual environment source venv/bin/activate # Install dependencies if not already installed pip install -r requirements.txt # Run the application python3 main.py
Step 9: Testing All CRUD Operations
When you run the application, you’ll see a menu with 11 options. Let’s test each CRUD operation:
1. GET all posts (Read)
- Select option 1
- Demonstrates retrieving a collection of resources
- Shows formatted table with post previews
2. GET a specific post (Read)
- Select option 2, then enter post ID (e.g., 1)
- Shows retrieving a single resource by ID
- Displays detailed post information
3. POST a new post (Create)
- Select option 3
- Enter user ID, title, and body content
- Shows JSON serialization and POST request
- Returns the created post with generated ID
4. PUT to update a post (Update)
- Select option 4
- Enter post ID to update
- Enter new user ID, title, and body
- Shows full resource replacement
5. PATCH to partially update (Update)
- Select option 5
- Enter post ID to update
- Choose which fields to update
- Shows partial resource updates
6. DELETE a post (Delete)
- Select option 6
- Enter post ID to delete
- Shows resource deletion
- Note: JSONPlaceholder is fake, so deletion is simulated
7-10. Additional Operations
- Options 7-10 demonstrate working with users and search
- Shows how to handle related resources
Key Concepts Demonstrated
1. HTTP Client Implementation
- Uses
requestslibrary for HTTP operations - Includes timeout handling
- Has proper error handling for network issues
2. Data Modeling
- Uses Python dataclasses for clean data models
- Handles JSON serialization/deserialization
- Supports both camelCase (API) and snake_case (Python)
3. Separation of Concerns
- api_client.py: Pure API communication
- models.py: Data structures
- utils.py: Display and input utilities
- main.py: Application logic and user interface
4. Error Handling
- Network errors (timeout, connection)
- HTTP errors (404, 500, etc.)
- JSON parsing errors
- User input validation
5. User Experience
- Color-coded output for better readability
- Formatted tables for data display
- Interactive prompts with validation
- Clear menu navigation
Troubleshooting
Common Issues:
- Import errors:
pip install -r requirements.txt
- Network connectivity:
- Check internet connection
- Verify API URL is accessible:
https://jsonplaceholder.typicode.com
- Python version:
python3 --version # Should be 3.7+
- Permission issues:
chmod +x run.sh
Extending the Project
This project can be extended in several ways:
- Add more resources: Extend to comments, todos, albums, etc.
- Add caching: Implement caching for frequently accessed data
- Add configuration: Make API URL configurable
- Add logging: Implement comprehensive logging
- Add unit tests: Test each component
- Add export functionality: Export data to CSV/JSON files
Conclusion
You’ve successfully built a complete Python command-line application that performs all CRUD operations against a REST API! This project demonstrates:
- Professional Python project structure
- REST API consumption with proper error handling
- Clean separation of concerns
- Interactive command-line interface
- Comprehensive CRUD operations
The skills learned here are directly applicable to real-world API integrations and can be extended to work with any REST API by modifying the models and endpoints.
Final Project Structure:
python-api-client/ ├── api_client.py ├── models.py ├── main.py ├── utils.py ├── requirements.txt ├── README.md └── run.sh
You now have a solid foundation for consuming REST APIs in Python and can adapt this template for any API integration needs.
Leave a Comment
Your email address will not be published. Required fields are marked with *