API Analytics & Usage Tracking

# FastAPI middleware for tracking API usage

from fastapi import FastAPI, Request
import time
from datetime import datetime, timezone
import asyncpg
import json

app = FastAPI()

# Database pool for analytics (PostgreSQL/TimescaleDB)
async def get_analytics_pool():
    return await asyncpg.create_pool(
        "postgresql://analytics_db:5432/api_metrics"
    )

@app.middleware("http")
async def track_api_usage(request: Request, call_next):
    """
    Track every API request: endpoint, method, latency, status, user.
    Store in TimescaleDB for time-series analytics.
    """
    start_time = time.time()

    # Extract user/tenant from JWT or API key
    user_id = request.state.user_id if hasattr(request.state, 'user_id') else None
    tenant_id = request.state.tenant_id if hasattr(request.state, 'tenant_id') else None

    # Process request
    response = await call_next(request)

    # Calculate latency
    latency_ms = (time.time() - start_time) * 1000

    # Extract request/response sizes
    request_size = int(request.headers.get('content-length', 0))
    response_size = int(response.headers.get('content-length', 0))

    # Prepare metrics
    metrics = {
        'timestamp': datetime.now(timezone.utc),
        'endpoint': request.url.path,
        'method': request.method,
        'status_code': response.status_code,
        'latency_ms': latency_ms,
        'user_id': user_id,
        'tenant_id': tenant_id,
        'request_size': request_size,
        'response_size': response_size,
        'api_version': request.headers.get('X-API-Version', 'v1'),
        'region': 'us-east-1',  # Get from env or request header
    }

    # Async write to analytics DB (non-blocking)
    pool = await get_analytics_pool()
    await pool.execute("""
        INSERT INTO api_requests (
            timestamp, endpoint, method, status_code, latency_ms,
            user_id, tenant_id, request_size, response_size,
            api_version, region
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
    """, *metrics.values())

    # Add custom header with request ID for tracing
    response.headers['X-Request-ID'] = str(request.state.request_id)

    return response


# Alternative: Send metrics to external service (async, don't block response)
import httpx

async def send_to_analytics_service(metrics):
    """Send metrics to external analytics platform (Segment, Mixpanel, custom)"""
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://analytics.example.com/api/track",
            json=metrics,
            headers={"Authorization": f"Bearer {ANALYTICS_API_KEY}"}
        )

# Alternative: Log to stdout, ingest with Fluentd/Logstash
import logging
logger = logging.getLogger("api_metrics")

def log_metrics(metrics):
    """Structured logging for ingestion by log aggregators"""
    logger.info(
        "api_request",
        extra={"metrics": metrics}  # Structured log for parsing
    )

# Usage metering with Redis

from fastapi import FastAPI, HTTPException, Request, Depends
import redis.asyncio as redis
from datetime import datetime, timezone
import calendar

app = FastAPI()

# Redis connection for usage counters
redis_client = redis.from_url("redis://localhost:6379")

async def get_current_month():
    """Get current month as YYYY-MM for billing period"""
    now = datetime.now(timezone.utc)
    return f"{now.year}-{now.month:02d}"

async def increment_usage(tenant_id: str, endpoint: str, cost: float = 1.0):
    """
    Increment usage counter for tenant.
    Store in Redis for fast access, flush to DB for billing.
    """
    month = await get_current_month()

    # Increment monthly total
    total_key = f"usage:{tenant_id}:{month}:total"
    await redis_client.incrbyfloat(total_key, cost)

    # Increment per-endpoint counter
    endpoint_key = f"usage:{tenant_id}:{month}:endpoint:{endpoint}"
    await redis_client.incr(endpoint_key)

    # Set TTL (expire after 3 months)
    await redis_client.expire(total_key, 7776000)  # 90 days

    # Return current usage
    total_usage = float(await redis_client.get(total_key) or 0)
    return total_usage

async def get_usage(tenant_id: str, month: str = None):
    """Get usage for tenant in given month"""
    if not month:
        month = await get_current_month()

    total_key = f"usage:{tenant_id}:{month}:total"
    usage = await redis_client.get(total_key)
    return float(usage) if usage else 0.0

async def check_quota(tenant_id: str):
    """
    Check if tenant has exceeded quota.
    Return quota info and whether request should be allowed.
    """
    # Get tenant's plan (from database)
    plan = await get_tenant_plan(tenant_id)  # {"quota": 10000, "overage_allowed": True}

    # Get current usage
    current_usage = await get_usage(tenant_id)

    # Check quota
    if current_usage >= plan["quota"]:
        if not plan["overage_allowed"]:
            raise HTTPException(
                status_code=429,
                detail={
                    "error": "Quota exceeded",
                    "quota": plan["quota"],
                    "usage": current_usage,
                    "reset_at": get_next_month_start()
                }
            )
        # Allow overage but flag for billing

    return {
        "allowed": True,
        "quota": plan["quota"],
        "usage": current_usage,
        "remaining": max(0, plan["quota"] - current_usage)
    }


# Middleware to track usage and enforce quotas
@app.middleware("http")
async def meter_and_enforce_quota(request: Request, call_next):
    tenant_id = request.state.tenant_id

    # Check quota before processing
    quota_info = await check_quota(tenant_id)

    # Determine cost (different endpoints have different costs)
    endpoint_costs = {
        "/api/users": 1.0,       # 1 unit
        "/api/orders": 10.0,     # 10 units (expensive operation)
        "/api/analytics": 5.0,   # 5 units
    }
    cost = endpoint_costs.get(request.url.path, 1.0)

    # Process request
    response = await call_next(request)

    # Increment usage (only for successful requests)
    if 200 <= response.status_code < 300:
        new_usage = await increment_usage(tenant_id, request.url.path, cost)

        # Add usage headers to response
        response.headers['X-Usage-Quota'] = str(quota_info['quota'])
        response.headers['X-Usage-Current'] = str(int(new_usage))
        response.headers['X-Usage-Remaining'] = str(max(0, quota_info['quota'] - int(new_usage)))

    return response


# Endpoint to get usage stats
@app.get("/api/usage")
async def get_usage_stats(tenant_id: str = Depends(get_current_tenant)):
    """Return usage statistics for current billing period"""
    month = await get_current_month()
    total_usage = await get_usage(tenant_id, month)

    # Get per-endpoint breakdown
    endpoint_pattern = f"usage:{tenant_id}:{month}:endpoint:*"
    endpoint_keys = await redis_client.keys(endpoint_pattern)

    endpoints = {}
    for key in endpoint_keys:
        endpoint = key.decode().split(":")[-1]
        count = int(await redis_client.get(key))
        endpoints[endpoint] = count

    return {
        "tenant_id": tenant_id,
        "billing_period": month,
        "total_usage": total_usage,
        "endpoints": endpoints,
        "quota": (await get_tenant_plan(tenant_id))["quota"]
    }

# Report usage to Stripe for billing

from datetime import datetime, timezone
import stripe
stripe.api_key = "sk_live_..."

# Create usage-based price in Stripe (one-time setup)
price = stripe.Price.create(
    product="prod_APIRequests",
    currency="usd",
    recurring={
        "interval": "month",
        "usage_type": "metered",  # Usage-based
        "aggregate_usage": "sum"  # Sum all usage events
    },
    billing_scheme="per_unit",
    unit_amount_decimal="0.01"  # $0.01 per request
)

# Subscribe customer to usage-based plan
subscription = stripe.Subscription.create(
    customer="cus_ABC123",
    items=[{"price": price.id}]
)

# Report usage events (call this from your API middleware)
def report_usage_to_stripe(tenant_id: str, quantity: int = 1):
    """Report API usage to Stripe for billing"""

    # Get Stripe subscription ID for tenant
    subscription_id = get_stripe_subscription_id(tenant_id)
    subscription_item_id = get_subscription_item_id(subscription_id)

    # Report usage
    stripe.SubscriptionItem.create_usage_record(
        subscription_item_id,
        quantity=quantity,
        timestamp=int(datetime.now(timezone.utc).timestamp()),
        action="increment"  # or "set" to replace
    )

# Batch reporting (more efficient for high volume)
async def batch_report_usage():
    """
    Periodically (every 5 minutes) flush usage from Redis to Stripe.
    Reduces API calls to Stripe.
    """
    for tenant_id in get_all_tenants():
        # Get usage from Redis since last flush
        usage_key = f"usage_pending:{tenant_id}"
        pending_usage = int(await redis_client.get(usage_key) or 0)

        if pending_usage > 0:
            # Report to Stripe
            report_usage_to_stripe(tenant_id, pending_usage)

            # Reset pending counter
            await redis_client.set(usage_key, 0)

# At month-end, Stripe automatically:
# 1. Aggregates all usage events
# 2. Calculates bill (quantity × price)
# 3. Creates invoice
# 4. Charges customer
# 5. Sends invoice email

# API endpoint for dashboard data

from fastapi import FastAPI, Depends
from datetime import datetime, timedelta, timezone

@app.get("/api/analytics/overview")
async def get_analytics_overview(
    tenant_id: str = Depends(get_current_tenant),
    days: int = 30
):
    """
    Return analytics overview for customer dashboard.
    """
    start_date = datetime.now(timezone.utc) - timedelta(days=days)

    # Query TimescaleDB for aggregated metrics
    query = """
        SELECT
            date_trunc('day', timestamp) as day,
            COUNT(*) as total_requests,
            COUNT(*) FILTER (WHERE status_code >= 500) as error_count,
            AVG(latency_ms) as avg_latency,
            PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) as p95_latency
        FROM api_requests
        WHERE tenant_id = $1 AND timestamp >= $2
        GROUP BY day
        ORDER BY day
    """

    daily_stats = await db.fetch(query, tenant_id, start_date)

    # Get endpoint breakdown
    endpoint_query = """
        SELECT endpoint, COUNT(*) as count
        FROM api_requests
        WHERE tenant_id = $1 AND timestamp >= $2
        GROUP BY endpoint
        ORDER BY count DESC
        LIMIT 10
    """
    top_endpoints = await db.fetch(endpoint_query, tenant_id, start_date)

    # Get current quota info
    quota_info = await check_quota(tenant_id)

    return {
        "daily_usage": [
            {
                "date": str(row["day"].date()),
                "requests": row["total_requests"],
                "errors": row["error_count"],
                "avg_latency": float(row["avg_latency"]),
                "p95_latency": float(row["p95_latency"])
            }
            for row in daily_stats
        ],
        "top_endpoints": [
            {"endpoint": row["endpoint"], "count": row["count"]}
            for row in top_endpoints
        ],
        "quota": {
            "limit": quota_info["quota"],
            "used": quota_info["usage"],
            "remaining": quota_info["remaining"],
            "percent_used": (quota_info["usage"] / quota_info["quota"]) * 100
        },
        "period": {
            "start": str(start_date.date()),
            "end": str(datetime.now(timezone.utc).date()),
            "days": days
        }
    }


# React component to display dashboard
# Using Recharts library for visualizations
#
# import { LineChart, Line, XAxis, YAxis, Tooltip } from 'recharts';
#
# function UsageDashboard() {
#   const [data, setData] = useState(null);
#
#   useEffect(() => {
#     fetch('/api/analytics/overview')
#       .then(res => res.json())
#       .then(setData);
#   }, []);
#
#   return (
#     <div>
#       <h2>API Usage Dashboard</h2>
#
#       {/* Quota Gauge */}
#       <div className="quota-card">
#         <h3>Monthly Quota</h3>
#         <CircularProgress value={data.quota.percent_used} />
#         <p>{data.quota.used} / {data.quota.limit} requests</p>
#       </div>
#
#       {/* Usage Chart */}
#       <LineChart data={data.daily_usage}>
#         <Line dataKey="requests" stroke="#17724d" />
#         <XAxis dataKey="date" />
#         <YAxis />
#         <Tooltip />
#       </LineChart>
#
#       {/* Top Endpoints */}
#       <BarChart data={data.top_endpoints}>
#         <Bar dataKey="count" fill="#17724d" />
#       </BarChart>
#     </div>
#   );
# }

# SQL query to identify power users

SELECT
    tenant_id,
    tenant_name,
    COUNT(*) as total_requests,
    COUNT(DISTINCT DATE(timestamp)) as active_days,
    SUM(CASE WHEN status_code >= 500 THEN 1 ELSE 0 END) as error_count,
    AVG(latency_ms) as avg_latency,
    SUM(request_size + response_size) / 1024 / 1024 as total_mb,
    -- Calculate revenue (example: $0.01 per request)
    COUNT(*) * 0.01 as estimated_revenue
FROM api_requests
WHERE timestamp >= NOW() - INTERVAL '30 days'
GROUP BY tenant_id, tenant_name
HAVING COUNT(*) > 100000  -- More than 100k requests
ORDER BY total_requests DESC
LIMIT 100;

-- Power user scoring
-- Combine multiple signals into a score

WITH user_stats AS (
    SELECT
        tenant_id,
        COUNT(*) as requests,
        COUNT(DISTINCT DATE(timestamp)) as active_days,
        AVG(latency_ms) as avg_latency
    FROM api_requests
    WHERE timestamp >= NOW() - INTERVAL '30 days'
    GROUP BY tenant_id
)
SELECT
    tenant_id,
    -- Normalize and weight different signals
    (
        (requests / (SELECT MAX(requests) FROM user_stats)) * 0.4 +  -- 40% weight
        (active_days / 30.0) * 0.3 +                                  -- 30% weight
        (1 - avg_latency / 1000.0) * 0.1 +                           -- 10% weight (faster = better)
        -- Add more signals...
    ) * 100 as power_user_score
FROM user_stats
WHERE requests > 1000  -- Minimum threshold
ORDER BY power_user_score DESC;

# Track costs per tenant

from fastapi import FastAPI, Request
import asyncpg
from datetime import datetime, timezone

# Cost constants (adjust based on your infrastructure)
COST_PER_CPU_SECOND = 0.00001  # AWS Lambda pricing
COST_PER_GB_EGRESS = 0.09       # AWS data transfer out
COST_PER_DB_QUERY = 0.0001      # Estimated DB cost

@app.middleware("http")
async def track_costs(request: Request, call_next):
    """
    Attribute infrastructure costs to tenant for profitability analysis.
    """
    start_time = time.time()
    tenant_id = request.state.tenant_id

    # Process request
    response = await call_next(request)

    # Calculate costs
    latency_seconds = time.time() - start_time
    compute_cost = latency_seconds * COST_PER_CPU_SECOND

    response_size_gb = int(response.headers.get('content-length', 0)) / (1024**3)
    egress_cost = response_size_gb * COST_PER_GB_EGRESS

    # Estimate DB cost (track queries via DB middleware)
    db_queries = getattr(request.state, 'db_queries', 0)
    db_cost = db_queries * COST_PER_DB_QUERY

    total_cost = compute_cost + egress_cost + db_cost

    # Store cost attribution
    await db.execute("""
        INSERT INTO tenant_costs (
            tenant_id, timestamp, compute_cost, egress_cost,
            db_cost, total_cost, request_count
        ) VALUES ($1, $2, $3, $4, $5, $6, 1)
    """, tenant_id, datetime.now(timezone.utc), compute_cost, egress_cost,
         db_cost, total_cost)

    return response


# Daily cost aggregation and profitability report
async def calculate_tenant_profitability():
    """
    Compare costs vs revenue per tenant.
    Identify unprofitable customers.
    """
    query = """
        WITH costs AS (
            SELECT
                tenant_id,
                SUM(total_cost) as monthly_cost,
                SUM(request_count) as total_requests
            FROM tenant_costs
            WHERE timestamp >= NOW() - INTERVAL '30 days'
            GROUP BY tenant_id
        ),
        revenue AS (
            SELECT
                tenant_id,
                SUM(amount) as monthly_revenue
            FROM invoices
            WHERE created_at >= NOW() - INTERVAL '30 days'
            GROUP BY tenant_id
        )
        SELECT
            c.tenant_id,
            t.name as tenant_name,
            c.monthly_cost,
            COALESCE(r.monthly_revenue, 0) as monthly_revenue,
            COALESCE(r.monthly_revenue, 0) - c.monthly_cost as profit,
            c.total_requests,
            c.monthly_cost / NULLIF(c.total_requests, 0) as cost_per_request
        FROM costs c
        LEFT JOIN revenue r ON c.tenant_id = r.tenant_id
        LEFT JOIN tenants t ON c.tenant_id = t.id
        ORDER BY profit ASC  -- Most unprofitable first
    """

    results = await db.fetch(query)

    # Identify unprofitable tenants
    unprofitable = [
        row for row in results
        if row['profit'] < 0
    ]

    if unprofitable:
        print(f"⚠️  {len(unprofitable)} unprofitable tenants:")
        for row in unprofitable[:10]:  # Top 10
            print(f"  - {row['tenant_name']}: "
                  f"Revenue ${row['monthly_revenue']:.2f}, "
                  f"Cost ${row['monthly_cost']:.2f}, "
                  f"Loss ${abs(row['profit']):.2f}")

    return results


# Action items based on profitability
async def optimize_unprofitable_tenants():
    """
    Take action on unprofitable tenants.
    """
    unprofitable = await calculate_tenant_profitability()

    for tenant in unprofitable:
        if tenant['profit'] < -100:  # Losing $100+/month
            # Option 1: Reach out about upgrading plan
            await send_email(
                tenant['tenant_id'],
                subject="Let's discuss your API usage",
                body=f"Your usage ({tenant['total_requests']} requests) "
                     f"exceeds your current plan. We'd love to offer you "
                     f"an enterprise plan with better pricing."
            )

            # Option 2: Implement rate limiting if abuse
            if tenant['cost_per_request'] > 0.01:  # Abnormally expensive
                await apply_rate_limit(tenant['tenant_id'], 100)  # req/min

            # Option 3: Optimize their workload
            # Analyze which endpoints they use, offer caching, etc.