SaaS Usage Metering

import requests
import hashlib
import json
from datetime import datetime, timezone

API_BASE = "https://balance-api.pacspace.io"
API_KEY = "pk_live_PUBLIC.SECRET"  # Use env vars in production

HEADERS = {
    "X-Api-Key": API_KEY,
    "Content-Type": "application/json"
}\n\ndef record_api_call(customer_id: str, endpoint: str, response_time_ms: int):
    """
    Record a single API call as a verified usage delta.
    Call this from your API gateway or middleware.
    """
    record_key = f"usage-{customer_id}-{datetime.now(timezone.utc).strftime('%Y-%m')}"

    payload = {
        "items": [
            {
                "key": record_key,
                "accountId": customer_id,
                "bits": 1,  # Each API call = 1 unit
                "metadata": {
                    "endpoint": endpoint,
                    "responseTimeMs": response_time_ms,
                    "timestamp": datetime.now(timezone.utc).isoformat()
                }
            }
        ]
    }

    response = requests.post(f"{API_BASE}/api/v1/writes", headers=HEADERS, json=payload)
    response.raise_for_status()

    result = response.json()
    print(f"Recorded API call for {customer_id}: {endpoint} → batch {result['data']['batchId']}")
    return result["data"]\n\ndef get_usage(customer_id: str, month: str = None):
    """
    Get the total verified usage for a customer in a given month.
    month format: '2025-06'
    """
    if month is None:
        month = datetime.now(timezone.utc).strftime("%Y-%m")

    record_key = f"usage-{customer_id}-{month}"

    response = requests.get(f"{API_BASE}/api/v1/reads/bucket/{record_key}", headers=HEADERS)

    if response.status_code == 404:
        return {"customer_id": customer_id, "month": month, "total_calls": 0, "verified": False}

    response.raise_for_status()
    data = response.json()["data"]

    return {
        "customer_id": customer_id,
        "month": month,
        "total_calls": data["currentBits"],
        "total_writes": data["totalWrites"],
        "verified": data["verified"],
        "last_activity": data["lastWriteAt"]
    }\n\ndef generate_invoice(customer_id: str, month: str, rate_per_call: float = 0.001):
    """
    Generate a verifiable invoice based on confirmed usage data.
    """
    usage = get_usage(customer_id, month)

    if usage["total_calls"] == 0:
        print(f"No usage recorded for {customer_id} in {month}")
        return None

    invoice = {
        "invoice_id": f"inv-{customer_id}-{month}",
        "customer_id": customer_id,
        "billing_period": month,
        "total_api_calls": usage["total_calls"],
        "rate_per_call": rate_per_call,
        "subtotal": round(usage["total_calls"] * rate_per_call, 2),
        "verified": usage["verified"],
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "verification_record": f"usage-{customer_id}-{month}"
    }

    print(f"\n{'='*50}")
    print(f"  INVOICE: {invoice['invoice_id']}")
    print(f"{'='*50}")
    print(f"  Customer:       {invoice['customer_id']}")
    print(f"  Period:         {invoice['billing_period']}")
    print(f"  API Calls:      {invoice['total_api_calls']:,}")
    print(f"  Rate:           ${invoice['rate_per_call']:.4f} / call")
    print(f"  Subtotal:       ${invoice['subtotal']:,.2f}")
    print(f"  Verified:       {'Yes' if invoice['verified'] else 'Pending'}")
    print(f"  Verify at:      GET /api/v1/reads/bucket/{invoice['verification_record']}")
    print(f"{'='*50}\n")

    return invoice\n\ndef record_as_fact(invoice: dict):
    """
    Record the finalized invoice as an immutable fact for audit purposes.
    """
    content = json.dumps(invoice, sort_keys=True)

    payload = {
        "content": content,
        "metadata": {
            "type": "invoice",
            "invoice_id": invoice["invoice_id"],
            "customer_id": invoice["customer_id"],
            "period": invoice["billing_period"]
        }
    }

    response = requests.post(f"{API_BASE}/api/v1/facts", headers=HEADERS, json=payload)
    response.raise_for_status()

    result = response.json()["data"]
    print(f"Invoice recorded as fact: {result['contentHash']}")
    return result\n\n# --- Run the full flow ---

if __name__ == "__main__":
    customer = "cust-acme-corp"
    month = "2025-06"

    # Simulate recording API calls (in production, call from your API middleware)
    print("Recording API usage...\n")
    record_api_call(customer, "/api/v1/search", 45)
    record_api_call(customer, "/api/v1/search", 38)
    record_api_call(customer, "/api/v1/analytics", 120)
    record_api_call(customer, "/api/v1/export", 230)
    record_api_call(customer, "/api/v1/search", 52)

    # Check usage
    print("\nChecking verified usage...")
    usage = get_usage(customer, month)
    print(f"Total API calls: {usage['total_calls']}")

    # Generate invoice
    invoice = generate_invoice(customer, month, rate_per_call=0.002)

    # Record invoice as a verified fact
    if invoice:
        record_as_fact(invoice)

usage-cust-acme-corp-2025-06

# Flask middleware example
@app.after_request
def track_usage(response):
    if hasattr(request, 'customer_id'):
        record_api_call(
            customer_id=request.customer_id,
            endpoint=request.path,
            response_time_ms=int(request.elapsed_ms)
        )
    return response

Recording API usage...

Recorded API call for cust-acme-corp: /api/v1/search → batch batch_001
Recorded API call for cust-acme-corp: /api/v1/search → batch batch_002
Recorded API call for cust-acme-corp: /api/v1/analytics → batch batch_003
Recorded API call for cust-acme-corp: /api/v1/export → batch batch_004
Recorded API call for cust-acme-corp: /api/v1/search → batch batch_005

Checking verified usage...
Total API calls: 5

==================================================
  INVOICE: inv-cust-acme-corp-2025-06
==================================================
  Customer:       cust-acme-corp
  Period:         2025-06
  API Calls:      5
  Rate:           $0.0020 / call
  Subtotal:       $0.01
  Verified:       Yes
  Verify at:      GET /api/v1/reads/bucket/usage-cust-acme-corp-2025-06
==================================================

Invoice recorded as fact: sha256_a1b2c3d4e5f6...