Module 23 - Working with APIs

APIs (Application Programming Interfaces) allow applications to communicate with each other. Python's requests library makes it easy to work with REST APIs.

---

1. Introduction to REST APIs

REST (Representational State Transfer) is an architectural style for web services.

HTTP Methods

Method	Purpose	Example
GET	Retrieve data	Get user list
POST	Create data	Create new user
PUT	Update (replace) data	Update user info
PATCH	Partial update	Update user email only
DELETE	Delete data	Delete user

Status Codes

Code	Meaning	Example
200	OK	Successful GET
201	Created	Successful POST
204	No Content	Successful DELETE
400	Bad Request	Invalid data
401	Unauthorized	Missing/invalid auth
404	Not Found	Resource doesn't exist
500	Server Error	Server-side error

---

2. The requests Library

Installation

pip install requests

2.1 GET Requests

import requests

# Simple GET request
response = requests.get('https://api.github.com')
print(response.status_code)  # 200
print(response.text)          # Response body as string
print(response.json())        # Response body as JSON

# GET with query parameters
params = {'q': 'python', 'sort': 'stars'}
response = requests.get('https://api.github.com/search/repositories', params=params)
data = response.json()
print(data['total_count'])

2.2 POST Requests

import requests

# POST with JSON data
url = 'https://jsonplaceholder.typicode.com/posts'
data = {
    'title': 'My Post',
    'body': 'This is the content',
    'userId': 1
}

response = requests.post(url, json=data)
print(response.status_code)  # 201
print(response.json())

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

2.3 PUT and PATCH Requests

import requests

# PUT - Replace entire resource
url = 'https://jsonplaceholder.typicode.com/posts/1'
data = {
    'id': 1,
    'title': 'Updated Title',
    'body': 'Updated content',
    'userId': 1
}
response = requests.put(url, json=data)

# PATCH - Partial update
data = {'title': 'New Title Only'}
response = requests.patch(url, json=data)
print(response.json())

2.4 DELETE Requests

import requests

url = 'https://jsonplaceholder.typicode.com/posts/1'
response = requests.delete(url)
print(response.status_code)  # 200 or 204

---

3. Headers and Authentication

3.1 Custom Headers

import requests

headers = {
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json'
}

response = requests.get('https://api.github.com', headers=headers)
print(response.json())

3.2 Bearer Token Authentication

import requests

token = 'your_access_token_here'
headers = {'Authorization': f'Bearer {token}'}

response = requests.get('https://api.example.com/protected', headers=headers)

3.3 Basic Authentication

import requests

# Method 1: Using auth parameter
response = requests.get(
    'https://api.example.com/users',
    auth=('username', 'password')
)

# Method 2: Using headers
import base64

credentials = base64.b64encode(b'username:password').decode('utf-8')
headers = {'Authorization': f'Basic {credentials}'}
response = requests.get('https://api.example.com/users', headers=headers)

3.4 API Key Authentication

import requests

# In query parameters
params = {'api_key': 'your_api_key_here'}
response = requests.get('https://api.example.com/data', params=params)

# In headers
headers = {'X-API-Key': 'your_api_key_here'}
response = requests.get('https://api.example.com/data', headers=headers)

---

4. Error Handling

4.1 Checking Status Codes

import requests

response = requests.get('https://api.github.com/invalid-endpoint')

if response.status_code == 200:
    print('Success!')
    data = response.json()
elif response.status_code == 404:
    print('Not found')
elif response.status_code == 500:
    print('Server error')
else:
    print(f'Error: {response.status_code}')

4.2 Using raise_for_status()

import requests

try:
    response = requests.get('https://api.github.com/invalid')
    response.raise_for_status()  # Raises exception for 4xx/5xx
    data = response.json()
except requests.exceptions.HTTPError as e:
    print(f'HTTP error occurred: {e}')
except requests.exceptions.ConnectionError:
    print('Connection error')
except requests.exceptions.Timeout:
    print('Request timed out')
except requests.exceptions.RequestException as e:
    print(f'Error: {e}')

4.3 Comprehensive Error Handling

import requests
from typing import Optional, Dict

def fetch_data(url: str) -> Optional[Dict]:
    """Fetch data with comprehensive error handling"""
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f'HTTP error: {e}')
        print(f'Response: {response.text}')
    except requests.exceptions.ConnectionError:
        print('Failed to connect to server')
    except requests.exceptions.Timeout:
        print('Request timed out')
    except requests.exceptions.JSONDecodeError:
        print('Invalid JSON response')
    except Exception as e:
        print(f'Unexpected error: {e}')
    
    return None

data = fetch_data('https://api.github.com/users/octocat')
if data:
    print(data)

---

5. Practical Examples

5.1 GitHub API Client

import requests
from typing import List, Dict

class GitHubAPI:
    BASE_URL = 'https://api.github.com'
    
    def __init__(self, token: str = None):
        self.session = requests.Session()
        if token:
            self.session.headers['Authorization'] = f'token {token}'
    
    def get_user(self, username: str) -> Dict:
        """Get user information"""
        response = self.session.get(f'{self.BASE_URL}/users/{username}')
        response.raise_for_status()
        return response.json()
    
    def get_repos(self, username: str) -> List[Dict]:
        """Get user's repositories"""
        response = self.session.get(f'{self.BASE_URL}/users/{username}/repos')
        response.raise_for_status()
        return response.json()
    
    def search_repositories(self, query: str, sort: str = 'stars') -> List[Dict]:
        """Search repositories"""
        params = {'q': query, 'sort': sort}
        response = self.session.get(f'{self.BASE_URL}/search/repositories', params=params)
        response.raise_for_status()
        return response.json()['items']

# Usage
api = GitHubAPI()
user = api.get_user('octocat')
print(f"User: {user['name']}")

repos = api.get_repos('octocat')
print(f"Repositories: {len(repos)}")

5.2 Weather API Client

import requests
from typing import Dict

class WeatherAPI:
    BASE_URL = 'https://api.openweathermap.org/data/2.5'
    
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def get_current_weather(self, city: str) -> Dict:
        """Get current weather for a city"""
        params = {
            'q': city,
            'appid': self.api_key,
            'units': 'metric'
        }
        response = requests.get(f'{self.BASE_URL}/weather', params=params)
        response.raise_for_status()
        return response.json()
    
    def get_forecast(self, city: str, days: int = 5) -> Dict:
        """Get weather forecast"""
        params = {
            'q': city,
            'appid': self.api_key,
            'units': 'metric',
            'cnt': days
        }
        response = requests.get(f'{self.BASE_URL}/forecast', params=params)
        response.raise_for_status()
        return response.json()

# Usage
# api = WeatherAPI('your_api_key')
# weather = api.get_current_weather('London')
# print(f"Temperature: {weather['main']['temp']}°C")

5.3 REST API CRUD Operations

import requests
from typing import List, Dict, Optional

class UserAPI:
    BASE_URL = 'https://jsonplaceholder.typicode.com'
    
    def get_users(self) -> List[Dict]:
        """Get all users"""
        response = requests.get(f'{self.BASE_URL}/users')
        response.raise_for_status()
        return response.json()
    
    def get_user(self, user_id: int) -> Dict:
        """Get user by ID"""
        response = requests.get(f'{self.BASE_URL}/users/{user_id}')
        response.raise_for_status()
        return response.json()
    
    def create_user(self, user_data: Dict) -> Dict:
        """Create new user"""
        response = requests.post(f'{self.BASE_URL}/users', json=user_data)
        response.raise_for_status()
        return response.json()
    
    def update_user(self, user_id: int, user_data: Dict) -> Dict:
        """Update user"""
        response = requests.put(f'{self.BASE_URL}/users/{user_id}', json=user_data)
        response.raise_for_status()
        return response.json()
    
    def delete_user(self, user_id: int) -> bool:
        """Delete user"""
        response = requests.delete(f'{self.BASE_URL}/users/{user_id}')
        response.raise_for_status()
        return True

# Usage
api = UserAPI()

# Get all users
users = api.get_users()
print(f"Total users: {len(users)}")

# Get specific user
user = api.get_user(1)
print(f"User: {user['name']}")

# Create user
new_user = api.create_user({
    'name': 'John Doe',
    'email': 'john@example.com'
})
print(f"Created user: {new_user}")

---

6. Advanced Features

6.1 Sessions (Connection Pooling)

import requests

# Reuse connection for multiple requests
session = requests.Session()
session.headers.update({'Authorization': 'Bearer token123'})

# All requests use the same session
response1 = session.get('https://api.example.com/endpoint1')
response2 = session.get('https://api.example.com/endpoint2')
response3 = session.get('https://api.example.com/endpoint3')

session.close()

6.2 Timeouts

import requests

# Connect timeout: 3s, Read timeout: 10s
try:
    response = requests.get('https://api.example.com', timeout=(3, 10))
except requests.exceptions.Timeout:
    print('Request timed out')

# Single timeout value (applies to both)
response = requests.get('https://api.example.com', timeout=5)

6.3 Retries

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

# Configure retry strategy
retry_strategy = Retry(
    total=3,                    # Total retries
    backoff_factor=1,           # Wait 1, 2, 4 seconds
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS"]
)

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

# Requests will automatically retry on failure
response = session.get('https://api.example.com')

6.4 Rate Limiting

import requests
import time
from functools import wraps

def rate_limit(calls_per_second):
    """Decorator for rate limiting"""
    min_interval = 1.0 / calls_per_second
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = min_interval - elapsed
            
            if wait_time > 0:
                time.sleep(wait_time)
            
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        
        return wrapper
    return decorator

@rate_limit(calls_per_second=2)
def fetch_data(url):
    return requests.get(url)

# Max 2 calls per second
for i in range(5):
    response = fetch_data(f'https://api.example.com/item/{i}')
    print(f'Fetched item {i}')

---

7. File Uploads and Downloads

7.1 Upload Files

import requests

# Upload single file
files = {'file': open('document.pdf', 'rb')}
response = requests.post('https://api.example.com/upload', files=files)

# Upload with additional data
files = {'file': open('image.jpg', 'rb')}
data = {'title': 'My Image', 'description': 'A photo'}
response = requests.post('https://api.example.com/upload', files=files, data=data)

# Multiple files
files = [
    ('file', open('file1.txt', 'rb')),
    ('file', open('file2.txt', 'rb'))
]
response = requests.post('https://api.example.com/upload', files=files)

7.2 Download Files

import requests

# Download file
url = 'https://example.com/file.pdf'
response = requests.get(url)

with open('downloaded_file.pdf', 'wb') as file:
    file.write(response.content)

# Stream large files
url = 'https://example.com/large_file.zip'
response = requests.get(url, stream=True)

with open('large_file.zip', 'wb') as file:
    for chunk in response.iter_content(chunk_size=8192):
        file.write(chunk)

print('Download complete')

---

8. Pagination

Handling Paginated Responses

import requests
from typing import List, Dict

def fetch_all_pages(base_url: str, per_page: int = 100) -> List[Dict]:
    """Fetch all pages from a paginated API"""
    all_data = []
    page = 1
    
    while True:
        params = {'page': page, 'per_page': per_page}
        response = requests.get(base_url, params=params)
        response.raise_for_status()
        
        data = response.json()
        if not data:
            break
        
        all_data.extend(data)
        page += 1
    
    return all_data

# Usage
# users = fetch_all_pages('https://api.example.com/users')

---

9. Best Practices

Create an API Wrapper Class

import requests
from typing import Dict, Optional

class APIClient:
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url
        self.session = requests.Session()
        
        if api_key:
            self.session.headers['Authorization'] = f'Bearer {api_key}'
        
        # Set default timeout
        self.session.request = lambda *args, **kwargs: \
            requests.Session.request(self.session, *args, timeout=10, **kwargs)
    
    def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict:
        """Make HTTP request with error handling"""
        url = f'{self.base_url}/{endpoint.lstrip("/")}'
        
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            print(f'HTTP error: {e}')
            raise
        except Exception as e:
            print(f'Error: {e}')
            raise
    
    def get(self, endpoint: str, **kwargs) -> Dict:
        return self._make_request('GET', endpoint, **kwargs)
    
    def post(self, endpoint: str, **kwargs) -> Dict:
        return self._make_request('POST', endpoint, **kwargs)
    
    def put(self, endpoint: str, **kwargs) -> Dict:
        return self._make_request('PUT', endpoint, **kwargs)
    
    def delete(self, endpoint: str, **kwargs) -> Dict:
        return self._make_request('DELETE', endpoint, **kwargs)

# Usage
api = APIClient('https://api.example.com', api_key='your_key')
data = api.get('/users/123')

Best Practices

✅ Use sessions for multiple requests
✅ Always set timeouts
✅ Implement retry logic for unreliable APIs
✅ Handle errors comprehensively
✅ Use environment variables for API keys
✅ Respect rate limits

---

Summary

✅ requests library simplifies HTTP communication
✅ Use GET, POST, PUT, PATCH, DELETE for CRUD operations
✅ Handle authentication with headers or auth parameter
✅ Always implement error handling and timeouts
✅ Use sessions for better performance
✅ Respect API rate limits

---

Next Steps

In Module 24, you'll learn:

• Database access with sqlite3
• SQL basics and queries
• CRUD operations in databases
• Connection management

---

Practice Exercises

1. Create a GitHub repository search tool using the GitHub API
2. Build a weather dashboard that fetches data from multiple cities
3. Implement a REST API client with automatic retry and rate limiting
4. Create a file downloader with progress tracking
5. Build a simple web scraper that respects robots.txt

Challenge

Create a comprehensive API testing framework that:

• Supports multiple authentication methods
• Implements automatic retry with exponential backoff
• Logs all requests and responses
• Validates response schemas
• Generates test reports
• Handles pagination automatically