Python APIs with Requests: Integrating External Services

import requests

# GET request
response = requests.get('https://api.github.com/users/octocat')
print(f'Status: {response.status_code}')
print(f'Data: {response.json()}')

# GET with parameters
params = {'q': 'python', 'page': 1, 'per_page': 10}
response = requests.get('https://api.github.com/search/repositories', params=params)

# GET with headers
headers = {'User-Agent': 'MyApp/1.0', 'Accept': 'application/json'}
response = requests.get('https://api.github.com/repos/python/cpython', headers=headers)

# POST request
data = {'title': 'New Issue', 'body': 'This is a test issue'}
response = requests.post('https://api.github.com/repos/owner/repo/issues', json=data, headers=headers)

# PUT request
data = {'title': 'Updated Title'}
response = requests.put('https://api.github.com/repos/owner/repo/issues/1', json=data)

# DELETE request
response = requests.delete('https://api.github.com/repos/owner/repo/issues/1')

# PATCH request
data = {'state': 'closed'}
response = requests.patch('https://api.github.com/repos/owner/repo/issues/1', json=data)

# Check response
if response.ok:
    print('Success!')
else:
    print(f'Failed: {response.status_code} - {response.text}')

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

# API Key in URL
response = requests.get('https://api.example.com/data?api_key=your-api-key')

# Basic Authentication
response = requests.get('https://api.example.com/private', auth=('username', 'password'))

# Bearer Token
headers = {'Authorization': 'Bearer your-token'}
response = requests.get('https://api.example.com/user', headers=headers)

# GitHub Token
github_token = 'ghp_your_token'
headers = {
    'Authorization': f'Bearer {github_token}',
    'Accept': 'application/vnd.github.v3+json'
}
response = requests.get('https://api.github.com/user/repos', headers=headers)

# AWS Signature (using boto3 for AWS is better)
# For other APIs needing custom auth, use requests-auth
from requests.auth import AuthBase

class CustomAuth(AuthBase):
    def __init__(self, token):
        self.token = token
    
    def __call__(self, r):
        r.headers['X-Custom-Token'] = self.token
        return r

response = requests.get('https://api.example.com/data', auth=CustomAuth('token123'))

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

# Basic error handling
try:
    response = requests.get('https://api.example.com/data', timeout=10)
    response.raise_for_status()  # Raises exception for 4xx/5xx
    data = response.json()
except requests.exceptions.Timeout:
    print('Request timed out')
except requests.exceptions.ConnectionError:
    print('Connection error')
except requests.exceptions.HTTPError as e:
    print(f'HTTP error: {e}')
except requests.exceptions.RequestException as e:
    print(f'Other error: {e}')

# Automatic retries with backoff
session = requests.Session()
retry_strategy = Retry(
    total=3,                      # Total retries
    backoff_factor=1,             # Wait 1, 2, 4 seconds between retries
    status_forcelist=[429, 500, 502, 503, 504],  # Retry on these status codes
    allowed_methods=['GET', 'POST', 'PUT']
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)

# Use the session
response = session.get('https://api.example.com/data')

# Custom retry with exponential backoff
def request_with_retry(url, max_retries=3, delay=1):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f'Attempt {attempt + 1} failed: {e}')
            if attempt == max_retries - 1:
                raise
            time.sleep(delay * (2 ** attempt))  # Exponential backoff
    return None

import time
from datetime import datetime

# Respect rate limits
class RateLimitedClient:
    def __init__(self, rate_limit=60, period=60):
        self.rate_limit = rate_limit
        self.period = period
        self.requests_made = 0
        self.period_start = datetime.now()
    
    def get(self, url, **kwargs):
        # Check rate limit
        now = datetime.now()
        elapsed = (now - self.period_start).total_seconds()
        
        if elapsed > self.period:
            # New period
            self.requests_made = 0
            self.period_start = now
        
        if self.requests_made >= self.rate_limit:
            wait_time = self.period - elapsed
            print(f'Rate limit reached. Waiting {wait_time:.1f} seconds...')
            time.sleep(wait_time)
            self.requests_made = 0
            self.period_start = datetime.now()
        
        self.requests_made += 1
        return requests.get(url, **kwargs)

# Pagination examples
# GitHub pagination (Link header)
def get_all_github_repos(username):
    repos = []
    url = f'https://api.github.com/users/{username}/repos?per_page=100'
    
    while url:
        response = requests.get(url)
        repos.extend(response.json())
        
        # Check for next page in Link header
        if 'next' in response.links:
            url = response.links['next']['url']
        else:
            url = None
    
    return repos

# Page-based pagination
def get_all_pages(base_url, params=None):
    if params is None:
        params = {}
    
    results = []
    page = 1
    
    while True:
        params['page'] = page
        response = requests.get(base_url, params=params)
        data = response.json()
        
        if not data:
            break
        
        results.extend(data)
        page += 1
    
    return results

# Cursor-based pagination
def paginate_with_cursor(base_url, limit=100):
    results = []
    cursor = None
    
    while True:
        params = {'limit': limit}
        if cursor:
            params['cursor'] = cursor
        
        response = requests.get(base_url, params=params)
        data = response.json()
        results.extend(data['items'])
        
        cursor = data.get('next_cursor')
        if not cursor:
            break
    
    return results

class DevOpsAPIClient:
    """Reusable API client for DevOps tools"""
    
    def __init__(self, base_url, api_key=None, token=None, timeout=30):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        
        # Set headers
        if api_key:
            self.session.headers.update({'X-API-Key': api_key})
        if token:
            self.session.headers.update({'Authorization': f'Bearer {token}'})
        
        # Add retries
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)
    
    def _request(self, method, endpoint, **kwargs):
        """Make API request with error handling"""
        url = f'{self.base_url}/{endpoint.lstrip("/")}'
        
        try:
            response = self.session.request(
                method, url, 
                timeout=self.timeout,
                **kwargs
            )
            response.raise_for_status()
            return response.json() if response.content else None
        except requests.exceptions.HTTPError as e:
            print(f'HTTP error: {e.response.status_code} - {e.response.text}')
            raise
        except requests.exceptions.RequestException as e:
            print(f'Request error: {e}')
            raise
    
    def get(self, endpoint, **kwargs):
        return self._request('GET', endpoint, **kwargs)
    
    def post(self, endpoint, data=None, **kwargs):
        return self._request('POST', endpoint, json=data, **kwargs)
    
    def put(self, endpoint, data=None, **kwargs):
        return self._request('PUT', endpoint, json=data, **kwargs)
    
    def delete(self, endpoint, **kwargs):
        return self._request('DELETE', endpoint, **kwargs)
    
    def get_paginated(self, endpoint, **kwargs):
        """Get all pages of results"""
        results = []
        params = kwargs.get('params', {})
        page = 1
        
        while True:
            params['page'] = page
            response = self._request('GET', endpoint, params=params)
            
            if not response:
                break
            
            if isinstance(response, list):
                results.extend(response)
            elif isinstance(response, dict) and 'items' in response:
                results.extend(response['items'])
            else:
                results.append(response)
            
            if len(response) < (params.get('per_page', 100)):
                break
            
            page += 1
        
        return results

# Example usage
client = DevOpsAPIClient(
    base_url='https://api.github.com',
    token='ghp_your_token'
)

# Get user info
user = client.get('/user')
print(f'User: {user["login"]}')

# Create an issue
issue = client.post(
    '/repos/owner/repo/issues',
    data={'title': 'Bug report', 'body': 'Description here'}
)

# Get all repos (paginated)
repos = client.get_paginated('/user/repos', params={'per_page': 100})
print(f'Total repos: {len(repos)}')

#!/usr/bin/env python3
"""
api_status_monitor.py - Monitor multiple API endpoints
"""

import requests
import json
import time
from datetime import datetime
from typing import Dict, List
from concurrent.futures import ThreadPoolExecutor, as_completed

class APIStatusMonitor:
    """Monitor multiple API endpoints with health checks"""
    
    def __init__(self, config_file=None):
        self.endpoints = self.load_config(config_file)
        self.results = []
    
    def load_config(self, config_file):
        """Load API endpoints from config"""
        if config_file:
            with open(config_file, 'r') as f:
                return json.load(f)['endpoints']
        
        # Default configuration
        return [
            {'name': 'GitHub API', 'url': 'https://api.github.com/health', 'method': 'GET', 'expected_status': 200},
            {'name': 'Google', 'url': 'https://www.google.com', 'method': 'GET', 'expected_status': 200},
            {'name': 'Local Service', 'url': 'http://localhost:8080/health', 'method': 'GET', 'expected_status': 200, 'timeout': 5}
        ]
    
    def check_endpoint(self, endpoint: Dict) -> Dict:
        """Check a single endpoint"""
        start_time = time.time()
        result = {
            'name': endpoint['name'],
            'url': endpoint['url'],
            'timestamp': datetime.now().isoformat(),
            'status': 'unknown',
            'response_time': None,
            'status_code': None,
            'error': None
        }
        
        try:
            method = endpoint.get('method', 'GET').upper()
            timeout = endpoint.get('timeout', 10)
            expected_status = endpoint.get('expected_status', 200)
            
            response = requests.request(
                method, 
                endpoint['url'],
                timeout=timeout
            )
            
            result['status_code'] = response.status_code
            result['response_time'] = round((time.time() - start_time) * 1000, 2)
            
            if response.status_code == expected_status:
                result['status'] = 'healthy'
            elif 400 <= response.status_code < 500:
                result['status'] = 'client_error'
            elif 500 <= response.status_code < 600:
                result['status'] = 'server_error'
            else:
                result['status'] = 'unexpected'
                
        except requests.exceptions.Timeout:
            result['status'] = 'timeout'
            result['error'] = f'Timeout after {timeout}s'
        except requests.exceptions.ConnectionError:
            result['status'] = 'connection_error'
            result['error'] = 'Failed to connect'
        except Exception as e:
            result['status'] = 'error'
            result['error'] = str(e)
        
        return result
    
    def check_all(self, parallel=True):
        """Check all endpoints"""
        self.results = []
        
        if parallel:
            # Check endpoints in parallel
            with ThreadPoolExecutor(max_workers=10) as executor:
                futures = {executor.submit(self.check_endpoint, ep): ep for ep in self.endpoints}
                for future in as_completed(futures):
                    self.results.append(future.result())
        else:
            # Sequential checks
            for endpoint in self.endpoints:
                self.results.append(self.check_endpoint(endpoint))
        
        return self.results
    
    def generate_report(self):
        """Generate status report"""
        if not self.results:
            self.check_all()
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {
                'total': len(self.results),
                'healthy': sum(1 for r in self.results if r['status'] == 'healthy'),
                'unhealthy': sum(1 for r in self.results if r['status'] != 'healthy'),
                'average_response': round(sum(r.get('response_time', 0) for r in self.results) / len(self.results), 2)
            },
            'endpoints': self.results
        }
        
        return report
    
    def print_report(self):
        """Print colorful report to console"""
        report = self.generate_report()
        
        print("\n" + "="*60)
        print(f"API STATUS REPORT - {report['timestamp']}")
        print("="*60)
        print(f"Total: {report['summary']['total']} | Healthy: {report['summary']['healthy']} | Unhealthy: {report['summary']['unhealthy']}")
        print(f"Average Response: {report['summary']['average_response']}ms")
        print("-"*60)
        
        for endpoint in report['endpoints']:
            status_color = {
                'healthy': '✅',
                'timeout': '⏰',
                'connection_error': '🔌',
                'client_error': '⚠️',
                'server_error': '🔥',
                'error': '❌'
            }.get(endpoint['status'], '❓')
            
            response_info = f"{endpoint['response_time']}ms" if endpoint['response_time'] else 'N/A'
            status_info = f"HTTP {endpoint['status_code']}" if endpoint['status_code'] else ''
            
            print(f"{status_color} {endpoint['name']:20} {endpoint['status']:15} {response_info:10} {status_info}")
            if endpoint.get('error'):
                print(f"   └─ Error: {endpoint['error']}")
        
        print("="*60 + "\n")
    
    def save_report(self, filepath='api_status_report.json'):
        """Save report to JSON file"""
        report = self.generate_report()
        with open(filepath, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"Report saved to {filepath}")
    
    def run_continuous(self, interval=60):
        """Run monitoring continuously"""
        print(f"Starting continuous monitoring (interval: {interval}s)")
        print("Press Ctrl+C to stop\n")
        
        try:
            while True:
                self.check_all()
                self.print_report()
                time.sleep(interval)
        except KeyboardInterrupt:
            print("\nMonitoring stopped")


def main():
    import argparse
    
    parser = argparse.ArgumentParser(description='API Status Monitor')
    parser.add_argument('--config', '-c', help='Config file with endpoints')
    parser.add_argument('--continuous', action='store_true', help='Run continuously')
    parser.add_argument('--interval', type=int, default=60, help='Check interval in seconds')
    parser.add_argument('--output', '-o', help='Save report to file')
    
    args = parser.parse_args()
    
    monitor = APIStatusMonitor(args.config)
    
    if args.continuous:
        monitor.run_continuous(args.interval)
    else:
        monitor.check_all()
        monitor.print_report()
        
        if args.output:
            monitor.save_report(args.output)

if __name__ == '__main__':
    main()