🌐 Understanding and Using APIs

#!/usr/bin/env python3
"""
Webhook Endpoint with Security
Flask application that receives and validates webhook POST requests
"""

from flask import Flask, request, jsonify
import hashlib
import hmac
import logging


# ==============================================================================
# CONFIGURATION
# ==============================================================================

WEBHOOK_SECRET = "your_secret_key_here"      # Shared secret (keep secure!)

app = Flask(__name__)

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ==============================================================================
# SECURITY FUNCTIONS
# ==============================================================================

def verify_signature(payload, signature):
    """
    Verify webhook signature using HMAC-SHA256
    
    Args:
        payload (bytes): Raw request body
        signature (str): Signature from request header
    
    Returns:
        bool: True if signature is valid
    """
    if not signature:
        return False
    
    # Calculate expected signature
    expected = hmac.new(
        WEBHOOK_SECRET.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    # Securely compare (prevents timing attacks)
    return hmac.compare_digest(signature, expected)


# ==============================================================================
# EVENT HANDLERS
# ==============================================================================

def handle_device_status(data):
    """Handle device status change event"""
    device_name = data.get('device_name')
    new_status = data.get('status')
    logger.info(f"Device {device_name} status changed to {new_status}")
    # Add your business logic here


def handle_alert(data):
    """Handle alert triggered event"""
    alert_type = data.get('alert_type')
    severity = data.get('severity')
    logger.warning(f"Alert: {alert_type} (Severity: {severity})")
    # Add your notification logic here


# ==============================================================================
# WEBHOOK ENDPOINT
# ==============================================================================

@app.route('/webhook', methods=['POST'])
def webhook_handler():
    """
    Main webhook endpoint that receives POST requests
    
    Returns:
        JSON response with status code
    """
    
    # Get signature from header
    signature = request.headers.get('X-Hub-Signature')
    
    # Verify webhook authenticity
    if not verify_signature(request.data, signature):
        logger.warning("Invalid signature")
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Parse JSON payload
    try:
        data = request.json
    except Exception as err:
        logger.error(f"Invalid JSON: {err}")
        return jsonify({'error': 'Invalid JSON'}), 400
    
    # Get event type
    event_type = request.headers.get('X-Event-Type', 'unknown')
    logger.info(f"Received: {event_type}")
    
    # Route to appropriate handler
    if event_type == 'device.status.changed':
        handle_device_status(data)
    elif event_type == 'alert.triggered':
        handle_alert(data)
    else:
        logger.warning(f"Unknown event: {event_type}")
    
    # Return success immediately
    return jsonify({'status': 'success', 'event_type': event_type}), 200


# ==============================================================================
# HEALTH CHECK ENDPOINT
# ==============================================================================

@app.route('/health', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({'status': 'healthy'}), 200


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    print("🚀 Starting webhook server...")
    print("📍 Webhook: http://localhost:5000/webhook")
    print("💚 Health: http://localhost:5000/health")
    
    app.run(host='0.0.0.0', port=5000, debug=True)

#!/usr/bin/env python3
"""
Register Webhook with API Service
Subscribe to webhook events by registering your endpoint URL
"""

import requests
import json


# ==============================================================================
# CONFIGURATION
# ==============================================================================

WEBHOOK_API_URL = "https://api.service.com/v1/webhooks"
YOUR_API_TOKEN = "<YOUR_BEARER_TOKEN>"
YOUR_WEBHOOK_URL = "https://your-domain.com/webhook"     # Must be publicly accessible
WEBHOOK_SECRET = "your_webhook_secret_here"


# ==============================================================================
# WEBHOOK REGISTRATION FUNCTION
# ==============================================================================

def register_webhook(webhook_url, events, secret):
    """
    Register webhook with API service
    
    Args:
        webhook_url (str): Your endpoint URL
        events (list): Event types to subscribe to
        secret (str): Shared secret for verification
    
    Returns:
        dict: Registration response or None
    """
    
    headers = {
        "Authorization": f"Bearer {YOUR_API_TOKEN}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "url": webhook_url,
        "events": events,
        "secret": secret
    }
    
    try:
        print(f"📤 Registering webhook...")
        print(f"🎯 URL: {webhook_url}")
        print(f"📋 Events: {', '.join(events)}")
        
        response = requests.post(
            WEBHOOK_API_URL,
            headers=headers,
            json=payload,
            timeout=10
        )
        
        response.raise_for_status()
        
        webhook_data = response.json()
        print(f"\n✅ Webhook registered!")
        print(f"🆔 ID: {webhook_data.get('id', 'N/A')}")
        print(json.dumps(webhook_data, indent=2))
        
        return webhook_data
        
    except requests.exceptions.HTTPError as err:
        print(f"❌ HTTP Error: {err}")
        
    except requests.exceptions.RequestException as err:
        print(f"⚠️ Request Error: {err}")
        
    except Exception as err:
        print(f"❌ Error: {err}")
    
    return None


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    events = ["device.created", "device.updated", "device.deleted"]
    
    result = register_webhook(YOUR_WEBHOOK_URL, events, WEBHOOK_SECRET)
    
    if result:
        print("\n✨ Webhook is now active!")
    else:
        print("\n⚠️ Registration failed!")

#!/usr/bin/env python3
"""
Idempotent Webhook Handler with Redis
Prevents duplicate processing of webhook events
"""

from flask import Flask, request, jsonify
import redis
import logging
from datetime import datetime


# ==============================================================================
# CONFIGURATION
# ==============================================================================

app = Flask(__name__)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Redis connection for event tracking
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

EVENT_TTL = 86400  # 24 hours in seconds


# ==============================================================================
# IDEMPOTENCY FUNCTIONS
# ==============================================================================

def is_event_processed(event_id):
    """Check if event was already processed"""
    key = f'webhook:{event_id}'
    return redis_client.exists(key) > 0


def mark_event_processed(event_id):
    """Mark event as processed in Redis with TTL"""
    key = f'webhook:{event_id}'
    value = datetime.now().isoformat()
    redis_client.setex(key, EVENT_TTL, value)
    logger.info(f"Marked event {event_id} as processed")


# ==============================================================================
# WEBHOOK ENDPOINT
# ==============================================================================

@app.route('/webhook', methods=['POST'])
def webhook():
    """Idempotent webhook handler - ensures events processed only once"""
    
    # Get unique event ID
    event_id = request.headers.get('X-Event-ID')
    
    if not event_id:
        return jsonify({'error': 'Missing X-Event-ID header'}), 400
    
    # Check if already processed
    if is_event_processed(event_id):
        logger.info(f"Event {event_id} already processed")
        return jsonify({'status': 'already_processed'}), 200
    
    # Mark as processed
    mark_event_processed(event_id)
    
    # Process webhook data
    try:
        data = request.json
        logger.info(f"Processing: {data}")
        # TODO: Add to message queue here
    except Exception as err:
        logger.error(f"Error: {err}")
    
    # Return success immediately
    return jsonify({'status': 'received', 'event_id': event_id}), 200


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    print("🔔 Webhook Server with Idempotency")
    print("📍 Endpoint: http://localhost:5000/webhook")
    print("🔄 Redis-based idempotency enabled\n")
    
    app.run(host='0.0.0.0', port=5000, debug=True)

#!/usr/bin/env python3
"""
Testing Webhooks Locally with ngrok
Step-by-step guide and Python code for webhook testing
"""

import requests


# ==============================================================================
# STEP 1: Install ngrok
# ==============================================================================
# Download: https://ngrok.com/download
# Or use package manager:
#   Windows: choco install ngrok
#   Mac: brew install ngrok
#   Linux: snap install ngrok


# ==============================================================================
# STEP 2: Run local webhook server (Terminal 1)
# ==============================================================================
# python webhook_server.py
# Server runs on: http://localhost:5000


# ==============================================================================
# STEP 3: Start ngrok tunnel (Terminal 2)
# ==============================================================================
# Command: ngrok http 5000
# Creates tunnel: public URL -> localhost:5000


# ==============================================================================
# STEP 4: Get public URL from ngrok output
# ==============================================================================
# ngrok displays:
# Forwarding: https://abc123.ngrok.io -> http://localhost:5000
# Use: https://abc123.ngrok.io/webhook


# ==============================================================================
# STEP 5: Register webhook using ngrok URL
# ==============================================================================

def register_ngrok_webhook(ngrok_url):
    """Register ngrok URL as webhook endpoint"""
    
    webhook_config = {
        "url": f"{ngrok_url}/webhook",
        "events": ["device.created", "device.updated"],
        "secret": "test-secret-123"
    }
    
    response = requests.post(
        "https://api.service.com/v1/webhooks",
        headers={"Authorization": "Bearer <YOUR_TOKEN>"},
        json=webhook_config
    )
    
    return response.json()


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    # Replace with your actual ngrok URL
    ngrok_url = "https://abc123def456.ngrok.io"
    
    print(f"🌐 Registering webhook: {ngrok_url}")
    
    result = register_ngrok_webhook(ngrok_url)
    
    print(f"✅ Registered!")
    print(f"📍 Receiving at: {ngrok_url}/webhook")
    print(f"💡 Keep ngrok running!")

# Common rate limit response headers returned by API

X-RateLimit-Limit: 100
# Total number of requests allowed in the current time window

X-RateLimit-Remaining: 75
# Number of requests remaining in current time window

X-RateLimit-Reset: 1640995200
# Unix timestamp when the rate limit will reset

Retry-After: 60
# Number of seconds to wait before making another request

#!/usr/bin/env python3
"""
Rate Limit Handler with Auto-Retry
Handles 429 errors and respects Retry-After headers
"""

import time
import requests
from requests.exceptions import HTTPError


# ==============================================================================
# API CALL WITH RETRY LOGIC
# ==============================================================================

def api_call_with_retry(url, headers, max_retries=3):
    """Make API call with automatic retry on rate limit"""
    
    for attempt in range(max_retries):
        try:
            # Make GET request
            response = requests.get(url, headers=headers)
            response.raise_for_status()              # Raise error if 4xx/5xx
            
            # Check remaining rate limit
            remaining = int(response.headers.get('X-RateLimit-Remaining', 1))
            
            if remaining < 10:
                print(f"⚠️ Warning: Only {remaining} requests remaining")
            
            return response.json()          # Return parsed response
            
        except HTTPError as err:
            if err.response.status_code == 429:         # Rate limited
                retry_after = int(err.response.headers.get('Retry-After', 60))
                print(f"⏱️ Rate limited. Waiting {retry_after}s (Attempt {attempt+1}/{max_retries})")
                time.sleep(retry_after)
            else:
                raise                           # Different error, re-raise
    
    raise Exception("Max retries exceeded")


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    url = "https://api.example.com/v1/devices"
    headers = {"Authorization": "Bearer <YOUR_TOKEN>"}
    
    try:
        data = api_call_with_retry(url, headers)
        print(f"✅ Success! Retrieved {len(data)} devices")
    except Exception as err:
        print(f"❌ Failed: {err}")

#!/usr/bin/env python3
"""
OAuth Token Management with Auto-Refresh
Automatically refreshes expired tokens
"""

import requests
import time


# ==============================================================================
# API CLIENT WITH TOKEN MANAGEMENT
# ==============================================================================

class APIClient:
    """API Client with automatic token refresh"""
    
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None                # Will be fetched on first use
        self.token_expiry = 0                   # Unix timestamp
    
    
    def get_token(self):
        """Get valid token (fetches new one if expired)"""
        
        # Check if current token is still valid
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token          # Use existing token
        
        # Request new token from OAuth endpoint
        response = requests.post(
            'https://api.example.com/oauth/token',
            data={
                'grant_type': 'client_credentials',
                'client_id': self.client_id,
                'client_secret': self.client_secret
            }
        )
        
        data = response.json()
        
        # Save new token and expiry
        self.access_token = data['access_token']
        self.token_expiry = time.time() + data['expires_in'] - 60  # 60s buffer
        
        print(f"🔑 New token acquired (expires in {data['expires_in']}s)")
        return self.access_token


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    # Create client
    client = APIClient(
        client_id="your_client_id",
        client_secret="your_client_secret"
    )
    
    # Get token (will fetch new one)
    token = client.get_token()
    print(f"Token: {token[:20]}...")
    
    # Get token again (will reuse existing)
    token2 = client.get_token()
    print(f"Same token reused: {token == token2}")

#!/usr/bin/env python3
"""
Pagination Handler
Retrieves all data from paginated API endpoints
"""

import requests


# ==============================================================================
# PAGINATION FUNCTION
# ==============================================================================

def get_all_devices(base_url, headers):
    """Fetch all devices across all pages"""
    
    all_devices = []                         # Accumulator for all devices
    page = 1                                    # Start at page 1
    per_page = 100                              # Max items per page
    
    while True:
        # Request current page
        response = requests.get(
            f"{base_url}/devices",
            params={'page': page, 'per_page': per_page},
            headers=headers
        )
        
        data = response.json()                 # Parse response
        devices = data['devices']              # Extract devices array
        
        all_devices.extend(devices)           # Add to collection
        print(f"📄 Page {page}: {len(devices)} devices")
        
        # Check if last page
        if len(devices) < per_page:           # Fewer items = last page
            break
        
        page += 1                               # Next page
    
    return all_devices


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    base_url = "https://api.example.com"
    headers = {"Authorization": "Bearer <YOUR_TOKEN>"}
    
    print("🔄 Fetching all devices with pagination...\n")
    devices = get_all_devices(base_url, headers)
    
    print(f"\n✅ Total devices retrieved: {len(devices)}")

#!/usr/bin/env python3
"""
Timeout Configuration Example
Demonstrates setting connection and read timeouts
"""

import requests


# ==============================================================================
# CONFIGURATION
# ==============================================================================

url = "https://api.example.com/v1/devices"
headers = {"Authorization": "Bearer <YOUR_TOKEN>"}


# ==============================================================================
# REQUEST WITH TIMEOUT
# ==============================================================================

try:
    response = requests.get(
        url,
        headers=headers,
        timeout=(10, 30)                       # (connect_timeout, read_timeout)
        # 10s to establish connection, 30s to receive response
    )
    
    print(f"✅ Status: {response.status_code}")
    print(response.json())
    
except requests.exceptions.Timeout:
    print("⏱️ Timeout: Server took too long to respond")
    
except requests.exceptions.ConnectionError:
    print("❌ Connection Error: Check network/URL")
    
except requests.exceptions.RequestException as err:
    print(f"⚠️ Error: {err}")

#!/usr/bin/env python3
"""
Resilient API Client with Retry, Caching, and Rate Limiting
Production-ready API client with all best practices
"""

import requests
import time
import logging
from cachetools import TTLCache


# ==============================================================================
# CONFIGURATION
# ==============================================================================

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ==============================================================================
# RESILIENT API CLIENT CLASS
# ==============================================================================

class ResilientAPIClient:
    """API Client with retry, caching, and rate limiting"""
    
    def __init__(self, base_url, api_key, cache_ttl=300, max_retries=3):
        """Initialize API client with configuration"""
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.max_retries = max_retries
        self.cache = TTLCache(maxsize=100, ttl=cache_ttl)
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })
    
    
    def _make_request(self, method, endpoint, **kwargs):
        """Make HTTP request with retry and rate limit handling"""
        url = f"{self.base_url}{endpoint}"
        backoff = 1
        
        for attempt in range(1, self.max_retries + 1):
            try:
                response = self.session.request(method, url, timeout=(10, 30), **kwargs)
                
                # Handle rate limiting
                if response.status_code == 429:
                    retry_after = int(response.headers.get('Retry-After', backoff))
                    logger.warning(f"Rate limited. Waiting {retry_after}s...")
                    time.sleep(retry_after)
                    backoff *= 2
                    continue
                
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.Timeout:
                if attempt == self.max_retries:
                    raise
                time.sleep(backoff)
                backoff *= 2
                
            except requests.exceptions.RequestException as err:
                logger.error(f"Request failed: {err}")
                raise
        
        raise Exception(f"Failed after {self.max_retries} attempts")
    
    
    def get(self, endpoint, use_cache=True, **kwargs):
        """Make GET request with optional caching"""
        if use_cache and endpoint in self.cache:
            logger.debug(f"Cache hit: {endpoint}")
            return self.cache[endpoint]
        
        result = self._make_request('GET', endpoint, **kwargs)
        
        if use_cache:
            self.cache[endpoint] = result
        
        return result
    
    
    def post(self, endpoint, data=None, **kwargs):
        """Make POST request"""
        return self._make_request('POST', endpoint, json=data, **kwargs)
    
    
    def patch(self, endpoint, data=None, **kwargs):
        """Make PATCH request"""
        return self._make_request('PATCH', endpoint, json=data, **kwargs)
    
    
    def delete(self, endpoint, **kwargs):
        """Make DELETE request"""
        return self._make_request('DELETE', endpoint, **kwargs)


# ==============================================================================
# MAIN EXECUTION
# ==============================================================================

if __name__ == "__main__":
    # Initialize client
    client = ResilientAPIClient(
        base_url="https://api.example.com",
        api_key="<YOUR_API_KEY>",
        cache_ttl=300,
        max_retries=3
    )
    
    try:
        # GET request with caching
        print("📡 Making GET request...")
        devices = client.get("/v1/devices")
        print(f"✅ Retrieved {len(devices)} devices")
        
        # Same request (from cache)
        devices_cached = client.get("/v1/devices")
        print(f"⚡ Cached: {len(devices_cached)} devices")
        
        # POST request
        new_device = client.post("/v1/devices", data={
            "name": "Router-10",
            "ip": "10.0.0.10"
        })
        print(f"✅ Created: {new_device.get('name')}")
        
    except Exception as err:
        print(f"❌ Error: {err}")
    
    print("\n✨ Demo completed!")