#!/usr/bin/env python3
"""Print subscription quotas / credit usage from GET /v1/subscription/quotas."""
import json
import os
import sys
import urllib.error
import urllib.request
URL = "https://api.bigdata.com/v1/subscription/quotas"
# Omitted from output (legacy / redundant with other metrics).
SKIP_IDS = frozenset({"text:units", "pages:units"})
def unwrap_subscription_body(parsed):
"""Return (subscription_payload, envelope).
The live API wraps the subscription in ``results`` alongside ``errors`` and
``metadata``. Older or direct responses may expose fields at the top level.
"""
if isinstance(parsed, dict) and isinstance(parsed.get("results"), dict):
return parsed["results"], parsed
return parsed, parsed
def money_cents(value):
if value is None:
return "—"
return f"${value / 100:,.2f}"
def fmt_remaining(value):
if value is None:
return "—"
if isinstance(value, str) and value.lower() == "unlimited":
return "unlimited"
if isinstance(value, (int, float)):
return money_cents(value)
return str(value)
def fmt_quota_scalar(value):
"""Format a quota usage or limit (int, float, or 'unlimited')."""
if value == "unlimited":
return "unlimited"
if isinstance(value, float) and not value.is_integer():
return f"{value:g}"
if isinstance(value, (int, float)):
n = int(value) if isinstance(value, float) else value
return f"{n:,}"
return str(value)
def remaining_cell(used, limit):
if limit == "unlimited":
return "∞"
if used is None or limit is None:
return "—"
try:
rem = float(limit) - float(used)
except (TypeError, ValueError):
return "—"
if abs(rem - round(rem)) < 1e-9:
return f"{int(round(rem)):,}"
return f"{rem:g}"
def row_from_unit(unit):
used = unit.get("units_usage")
limit = unit.get("units_limit")
return (
fmt_quota_scalar(used) if used is not None else "—",
fmt_quota_scalar(limit),
remaining_cell(used, limit),
)
def row_from_credit_unit(unit):
"""Table row for per-metric credit usage (amounts in cents); Used column only."""
used = unit.get("units_usage")
return (money_cents(used) if used is not None else "—",)
def print_table(rows, headers=("Metric", "Used", "Limit", "Remaining")):
"""Print a simple ASCII table (metric left-aligned; other columns right-aligned)."""
if not rows:
return
n = len(headers)
cols = [[headers[i]] + [str(r[i]) for r in rows] for i in range(n)]
widths = [max(len(s) for s in col) for col in cols]
sep = "+" + "".join(f"-{w * '-'}-+" for w in widths)
def fmt_row(cells):
parts = [f"| {cells[0]:<{widths[0]}}"]
for i in range(1, n):
parts.append(f" | {cells[i]:>{widths[i]}}")
parts.append(" |")
return "".join(parts)
print(sep)
print(fmt_row(headers))
print(sep)
for r in rows:
print(fmt_row(r))
print(sep)
def label_for(unit_id, unit, fallback):
if unit is None:
return fallback
return unit.get("display_name") or fallback
def print_units_grouped(quota_units, row_fn, *, credit_tables=False):
"""Print usage tables grouped by product area.
``row_fn(unit)`` returns a tuple: (used, limit, remaining) for quota, or (used) only
for credit tables.
"""
by_id = {u["id"]: u for u in quota_units if u.get("id")}
table_headers = (
("Metric", "Used") if credit_tables else ("Metric", "Used", "Limit", "Remaining")
)
def append_row(metric, unit):
vals = row_fn(unit)
return (metric,) + tuple(vals)
# --- Search Service ---
search_spec = [
("search:smart_unit_read", "Smart Search"),
("search:unit_read", "Fast Search"),
("search:batch_unit_read", "Batch Search"),
]
search_rows = []
for uid, short in search_spec:
u = by_id.get(uid)
if u is None:
continue
search_rows.append(append_row(short, u))
if search_rows:
print("Search Service (API Query Units):")
print_table(search_rows, headers=table_headers)
print()
# --- Research Agent ---
research_sections = [
(
"Base model",
[
"chat:input_tokens",
"chat:output_tokens",
"chat:cached_input_tokens",
],
),
(
"Pro model",
[
"chat:input_tokens_pro_model",
"chat:output_tokens_pro_model",
"chat:cached_input_tokens_pro_model",
],
),
(
"Web Searches",
["chat:external_search"],
),
]
any_research = False
for subsection, ids in research_sections:
rows = []
for uid in ids:
u = by_id.get(uid)
if u is None:
continue
name = label_for(uid, u, uid)
rows.append(append_row(name, u))
if rows:
if not any_research:
print("Research Agent:")
any_research = True
print(f" {subsection}")
print_table(rows, headers=table_headers)
print()
# --- Content Enrichment & Indexing ---
content_spec = [
("private_content:emails", "Emails uploaded"),
]
content_rows = []
for uid, short in content_spec:
u = by_id.get(uid)
if u is None:
continue
content_rows.append(append_row(short, u))
if content_rows:
print("Content Enrichment & Indexing:")
print_table(content_rows, headers=table_headers)
print()
# --- Anything else (except skipped) ---
known = {uid for uid, _ in search_spec}
known.update(uid for _, ids in research_sections for uid in ids)
known.update(uid for uid, _ in content_spec)
known.update(SKIP_IDS)
other_rows = []
for u in quota_units:
uid = u.get("id")
if not uid or uid in known or uid in SKIP_IDS:
continue
name = u.get("display_name") or uid
other_rows.append(append_row(name, u))
if other_rows:
print("Other:")
print_table(other_rows, headers=table_headers)
print()
def print_quota_grouped(quota_units):
print_units_grouped(quota_units, row_from_unit)
def main():
api_key = os.environ.get("BIGDATA_API_KEY")
if not api_key:
print("Set BIGDATA_API_KEY to your API key.", file=sys.stderr)
sys.exit(1)
req = urllib.request.Request(
URL,
headers={"X-API-KEY": api_key, "Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read().decode()
except urllib.error.HTTPError as e:
print(f"HTTP {e.code}: {e.read().decode()}", file=sys.stderr)
sys.exit(1)
parsed = json.loads(raw)
data, envelope = unwrap_subscription_body(parsed)
errors = envelope.get("errors") if isinstance(envelope, dict) else None
if errors:
print("API errors:", file=sys.stderr)
print(json.dumps(errors, indent=2), file=sys.stderr)
sub_type = data.get("subscription_type", "unknown")
print("=" * 72)
print(" Bigdata — subscription usage")
print("=" * 72)
print(f"Organization: {data.get('org_id', '—')}")
print(f"Subscription type: {sub_type}")
print()
if sub_type == "credits":
c = data.get("credits") or {}
print("Credits (organization total)")
print(f" Limit: {money_cents(c.get('limit'))}")
print(f" Used: {money_cents(c.get('usage'))}")
print(f" Remaining: {fmt_remaining(c.get('remaining'))}")
print()
for i, bill in enumerate(data.get("billing") or [], 1):
period = bill.get("period") or {}
start, end = period.get("start"), period.get("end")
print(f"Billing period {i}: {start} → {end}")
print(f" Credit usage (this period): {money_cents(bill.get('credit_usage'))}")
print()
bill_units = [
u
for u in (bill.get("units") or [])
if u.get("id") not in SKIP_IDS
]
print_units_grouped(bill_units, row_from_credit_unit, credit_tables=True)
elif sub_type == "quota":
units = [
u
for u in (data.get("quota_units") or [])
if u.get("id") not in SKIP_IDS
]
print_quota_grouped(units)
else:
print("Raw response:")
print(json.dumps(parsed, indent=2))
print("=" * 72)
if __name__ == "__main__":
main()
