HTTP API Documentation¶

import requests

response = requests.get('http://localhost:8000/health')
health = response.json()

if health['status'] == 'healthy':
    print("✓ All systems operational")
    print(f"  Active WebSocket connections: {health['websocket']['active_connections']}")
else:
    print("✗ System unhealthy")
    if health['database'] == 'unhealthy':
        print("  - Database is down")
    if health['redis'] == 'unhealthy':
        print("  - Redis is down")

import requests

response = requests.get(
    'http://localhost:8000/authors',
    headers={'Authorization': f'Bearer {token}'},
    params={'search': 'John'}
)

authors = response.json()
for author in authors:
    print(f"{author['id']}: {author['name']}")

import requests

response = requests.post(
    'http://localhost:8000/authors',
    headers={'Authorization': f'Bearer {token}'},
    json={'name': 'John Doe'}
)

if response.status_code == 201:
    author = response.json()
    print(f"Created author with ID: {author['id']}")

import requests

response = requests.put(
    'http://localhost:8000/authors/1',
    headers={'Authorization': f'Bearer {token}'},
    json={'name': 'Jane Doe'}
)

if response.status_code == 200:
    author = response.json()
    print(f"Updated author: {author['name']}")

import requests

response = requests.delete(
    'http://localhost:8000/authors/1',
    headers={'Authorization': f'Bearer {token}'}
)

if response.status_code == 204:
    print("Author deleted successfully")

import requests

def get_all_authors(token):
    """Fetch all authors using pagination."""
    all_authors = []
    page = 1

    while True:
        response = requests.get(
            'http://localhost:8000/authors/paginated',
            headers={'Authorization': f'Bearer {token}'},
            params={'page': page, 'per_page': 100}
        )

        data = response.json()
        all_authors.extend(data['items'])

        if page >= data['meta']['pages']:
            break

        page += 1

    return all_authors

import requests
from datetime import datetime, timedelta

# Get failed actions for specific user in last 24 hours
start_date = (datetime.now() - timedelta(days=1)).isoformat()

response = requests.get(
    'http://localhost:8000/audit-logs',
    headers={'Authorization': f'Bearer {admin_token}'},
    params={
        'username': 'john.doe',
        'outcome': 'error',
        'start_date': start_date
    }
)

logs = response.json()
for log in logs['items']:
    print(f"{log['timestamp']}: {log['action_type']} {log['resource']} - {log['outcome']}")

import requests

# Track user activity over time
response = requests.get(
    f'http://localhost:8000/audit-logs/user/{user_id}',
    headers={'Authorization': f'Bearer {admin_token}'},
    params={'per_page': 100}
)

logs = response.json()
print(f"Total actions: {logs['meta']['total']}")
print(f"Success rate: {sum(1 for log in logs['items'] if log['outcome'] == 'success') / len(logs['items']) * 100:.1f}%")

import requests

response = requests.get('http://localhost:8000/api/profiling/status')
status = response.json()

if not status['scalene_installed']:
    print("Scalene is not installed. Install with: pip install scalene")
elif not status['enabled']:
    print("Profiling is disabled. Enable with: PROFILING_ENABLED=true")
else:
    print(f"Profiling enabled. Reports saved to: {status['output_directory']}")

import requests
from datetime import datetime

response = requests.get('http://localhost:8000/api/profiling/reports')
data = response.json()

print(f"Total reports: {data['total_count']}")
for report in data['reports']:
    created = datetime.fromtimestamp(report['created_at'])
    size_mb = report['size_bytes'] / 1024 / 1024
    print(f"  {report['filename']} - {size_mb:.2f}MB - {created}")

import requests

# Download and save report
filename = "websocket_profile_20250123_143000.html"
response = requests.get(f'http://localhost:8000/api/profiling/reports/{filename}')

with open(filename, 'wb') as f:
    f.write(response.content)

print(f"Report saved to: {filename}")

import requests

# Delete old profiling reports
response = requests.get('http://localhost:8000/api/profiling/reports')
reports = response.json()['reports']

for report in reports:
    # Delete reports older than 7 days
    age_days = (time.time() - report['created_at']) / 86400
    if age_days > 7:
        delete_response = requests.delete(
            f"http://localhost:8000/api/profiling/reports/{report['filename']}"
        )
        print(f"Deleted: {report['filename']}")

from app.dependencies.permissions import require_roles

@router.get("/authors", dependencies=[Depends(require_roles("get-authors"))])
async def get_authors():
    ...

import requests
from datetime import datetime, timedelta

class APIClient:
    def __init__(self, base_url):
        self.base_url = base_url
        self.token = None
        self.token_expiry = None

    def login(self, username, password):
        response = requests.post(
            f'{self.base_url}/login',
            json={'username': username, 'password': password}
        )
        data = response.json()

        self.token = data['access_token']
        self.token_expiry = datetime.now() + timedelta(seconds=data['expires_in'])

    def _ensure_authenticated(self):
        if not self.token or datetime.now() >= self.token_expiry:
            raise Exception('Token expired or missing')

    def get_authors(self, **filters):
        self._ensure_authenticated()
        response = requests.get(
            f'{self.base_url}/authors',
            headers={'Authorization': f'Bearer {self.token}'},
            params=filters
        )
        response.raise_for_status()
        return response.json()

import requests
from requests.exceptions import RequestException

def safe_api_call(func):
    def wrapper(*args, **kwargs):
        try:
            response = func(*args, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                print("Authentication failed")
            elif e.response.status_code == 403:
                print("Permission denied")
            elif e.response.status_code == 429:
                print("Rate limit exceeded")
            else:
                print(f"HTTP error: {e}")
        except RequestException as e:
            print(f"Request failed: {e}")
        return None

    return wrapper

@safe_api_call
def get_authors(token):
    return requests.get(
        'http://localhost:8000/authors',
        headers={'Authorization': f'Bearer {token}'}
    )

def fetch_all_pages(token, endpoint, per_page=100):
    """
    Fetch all pages from a paginated endpoint.

    Args:
        token: Authentication token
        endpoint: API endpoint URL
        per_page: Items per page

    Yields:
        Individual items from all pages
    """
    page = 1

    while True:
        response = requests.get(
            endpoint,
            headers={'Authorization': f'Bearer {token}'},
            params={'page': page, 'per_page': per_page}
        )
        response.raise_for_status()

        data = response.json()

        for item in data['items']:
            yield item

        if page >= data['meta']['pages']:
            break

        page += 1

# Usage
for author in fetch_all_pages(token, 'http://localhost:8000/authors_paginated'):
    print(f"{author['id']}: {author['name']}")

import time
from functools import wraps

def rate_limit_handler(func):
    """Decorator to handle rate limiting with automatic retry."""

    @wraps(func)
    def wrapper(*args, **kwargs):
        max_retries = 3
        retry_count = 0

        while retry_count < max_retries:
            try:
                response = func(*args, **kwargs)

                # Check rate limit headers
                remaining = int(response.headers.get('X-RateLimit-Remaining', 0))

                if remaining < 5:
                    # Approaching limit, slow down
                    time.sleep(1)

                return response

            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 429:
                    retry_after = int(e.response.headers.get('Retry-After', 60))
                    print(f"Rate limited. Waiting {retry_after}s...")
                    time.sleep(retry_after)
                    retry_count += 1
                else:
                    raise

        raise Exception("Max retries exceeded for rate limiting")

    return wrapper

@rate_limit_handler
def get_authors_safe(token):
    return requests.get(
        'http://localhost:8000/authors',
        headers={'Authorization': f'Bearer {token}'}
    )

# Good: Paginated request
response = requests.get(
    'http://localhost:8000/authors_paginated',
    params={'page': 1, 'per_page': 50}
)

# Avoid: Non-paginated request for large datasets
response = requests.get('http://localhost:8000/authors')  # May return thousands

# Good: Filter on server
response = requests.get(
    'http://localhost:8000/authors',
    params={'name': 'Smith'}
)

# Avoid: Fetch all and filter on client
response = requests.get('http://localhost:8000/authors')
authors = [a for a in response.json() if 'Smith' in a['name']]

session = requests.Session()
session.headers.update({'Authorization': f'Bearer {token}'})

# Reuses underlying TCP connection
for i in range(10):
    response = session.get('http://localhost:8000/authors')

import requests
import time

def monitor_service(url, interval=30):
    """Monitor service health."""
    while True:
        try:
            response = requests.get(f'{url}/health', timeout=5)
            health = response.json()

            if health['status'] != 'healthy':
                alert(f"Service unhealthy: {health}")

        except Exception as e:
            alert(f"Health check failed: {e}")

        time.sleep(interval)