"""Cloud billing pull API — the External API's cloud-spend surface. GET /api/v1/open/cloud/billing/daily cloud.billing:read GET /api/v1/open/cloud/billing/monthly cloud.billing:read GET /api/v1/open/cloud/billing/breakdown cloud.billing:read GET /api/v1/open/cloud/billing/changes cloud.billing:read GET /api/v1/open/cloud/billing/tag-keys cloud.billing:read Sibling of ``open_cloud_router`` (inventory); same domain placement and the same credential plane. What each layer owns: this module validates and shapes, :mod:`app.domains.cmdb.open_billing_service` queries, and :mod:`app.domains.cmdb.open_visibility` decides which subscriptions the client may aggregate at all. Design contract (documented for integrators in ``docs/api/external-api.md`` §B.6): * **Read-only, one cost basis** — always the business pre-discount (list) figure. No ``measure`` parameter exists; see the service module for why a toggle here would leak the contract discount rate. * **Whitelist-scoped** — a client only ever sees the subscriptions its ``cloud_visibility`` grants. An unrestricted client still sees everything, which is why granting this scope to an unrestricted client is a deliberate admin decision. * **Window caps** — 366 days for the day series, 36 months for the month series, and a tighter 92 days when the service breakdown is requested (that multiplies the cell count by the provider's service cardinality). * **Every response is self-describing** — measure, window, ``data_through``, ``provisional_from``, ``excludes_unallocatable``, and any FX rows that had to be dropped. * **Audited** — each successful read writes one audit row attributed to the service principal. A dump of the company's cloud spend is exactly the kind of read that must be reconstructable after the fact. """ from __future__ import annotations import logging import uuid as uuid_mod from datetime import UTC, date, datetime, timedelta from fastapi import APIRouter, Depends, Query, Request from sqlalchemy.ext.asyncio import AsyncSession from app.api_clients.auth import ApiClientContext, require_api_client from app.api_clients.scopes import SCOPE_CLOUD_BILLING_READ from app.core.database import get_db from app.core.dependencies import get_real_ip from app.core.exceptions import ValidationError from app.core.response import success_response from app.domains.cmdb.open_billing_breakdown import ( DEFAULT_TOP, MAX_TOP, TAG_KEY_SCAN_LIMIT, TAG_PREFIX, breakdown, breakdown_meta, parse_dimensions, tag_keys, ) from app.domains.cmdb.open_billing_changes import ( DEFAULT_HORIZON_DAYS, MAX_HORIZON_DAYS, changed_cells, changes_meta, changes_rows, horizon_start, ) from app.domains.cmdb.open_billing_service import ( BillingQuery, BillingResult, daily_costs, monthly_costs, resolve_subscriptions, response_meta, ) from app.domains.cmdb.open_visibility import resolve_client_cloud_visibility logger = logging.getLogger(__name__) router = APIRouter( prefix="/api/v1/open/cloud/billing", tags=["open-cloud-billing"], ) _DEFAULT_PAGE_SIZE = 200 #: Deliberately above the 200-row cap the inventory endpoints use. A #: year of daily rows across a few hundred subscriptions is ~100k rows; #: at 200 per page that is 500+ calls, which collides with the 300 #: req/min client limit for no benefit — the rows are small and already #: aggregated. Documented in §A.5 as a billing-specific cap. _MAX_PAGE_SIZE = 1000 _MAX_DAILY_SPAN_DAYS = 366 _MAX_MONTHLY_SPAN_MONTHS = 36 #: Service breakdown multiplies rows by service cardinality (AWS alone #: reports 50+ services on a busy account), so it gets its own, much #: tighter window. _MAX_SERVICE_SPAN_DAYS = 92 #: A grouping aggregates every line item in the window in one pass, and #: the high-cardinality dimensions (service, resource) are exactly the #: ones a year-wide request is most tempting for. _MAX_BREAKDOWN_SPAN_DAYS = 92 _VALID_ENVIRONMENTS = {"dev", "staging", "prod"} _VALID_DAILY_GROUP_BY = {"subscription", "subscription,service"} _DAILY_ROW_EXAMPLE = { "date": "2026-08-01", "subscription_id": "b1c2d3e4-0000-0000-0000-000000000000", "external_id": "123456789012", "name": "RD Compute", "provider": "aws", "environment": "prod", "amount_usd": 812.4531, "currency": "USD", "is_provisional": False, "source": "cost_facts", } _DAILY_EXAMPLE = { "success": True, "data": [_DAILY_ROW_EXAMPLE], "meta": { "trace_id": "…", "page": 1, "page_size": 200, "total": 1, "measure": "list", "measure_note": ( "Business pre-discount (list) cost. Not the invoiced net amount — " "do not reconcile against a provider invoice." ), "currency": "USD", "window": {"start": "2026-07-16", "end": "2026-08-14"}, "data_through": "2026-08-13", "provisional_from": "2026-08-12", "excludes_unallocatable": True, "sources": ["cost_facts"], }, "errors": [], } _MONTHLY_EXAMPLE = { "success": True, "data": [{ "month": "2026-07", "subscription_id": "b1c2d3e4-0000-0000-0000-000000000000", "external_id": "123456789012", "name": "RD Compute", "provider": "aws", "environment": "prod", "amount_usd": 24_517.88, "currency": "USD", "month_is_partial": False, "source": "cost_facts", }], "meta": {"trace_id": "…", "page": 1, "page_size": 200, "total": 1}, "errors": [], } _BREAKDOWN_EXAMPLE = { "success": True, "data": [ {"key": "Amazon EC2", "label": "Amazon EC2", "amount_usd": 18_204.51, "share": 0.62}, {"key": "Amazon S3", "label": "Amazon S3", "amount_usd": 6_140.02, "share": 0.21}, ], "meta": { "trace_id": "…", "group_by": "service", "measure": "list", "currency": "USD", "window": {"start": "2026-07-01", "end": "2026-07-31"}, "total_usd": 29_361.24, "excludes_unallocatable": True, "sources": ["cost_facts"], }, "errors": [], } _CHANGES_EXAMPLE = { "success": True, "data": [{ "date": "2026-08-13", "subscription_id": "b1c2d3e4-0000-0000-0000-000000000000", "external_id": "123456789012", "name": "RD Compute", "provider": "aws", "environment": "prod", "last_ingested_at": "2026-08-14T02:11:07+00:00", }], "meta": { "trace_id": "…", "page": 1, "page_size": 200, "total": 1, "since": "2026-08-13T00:00:00+00:00", "horizon": "2026-07-14", "changed_cells": 1, "signals": ["billing_daily.fetched_at", "cost_facts.updated_at"], }, "errors": [], } _TAG_KEYS_EXAMPLE = { "success": True, "data": [ {"key": "usedby", "sampled_spend_usd": 1_011_082.0, "providers": ["alicloud_intl", "aws", "gcp"]}, {"key": "environment", "sampled_spend_usd": 562_931.0, "providers": ["alicloud_intl", "aws", "gcp"]}, ], "meta": { "trace_id": "…", "window": {"start": "2026-07-16", "end": "2026-08-14"}, "sampled_rows": 62_097, }, "errors": [], } def _example(payload: dict) -> dict: return {200: {"content": {"application/json": {"example": payload}}}} # ── Input validation ────────────────────────────────────────────────── def _resolve_day_window( start: date | None, end: date | None, *, max_days: int, narrow_hint: str = "Narrow the window.", ) -> tuple[date, date]: """Default to the trailing 30 days; validate order and span. ``end`` defaults to today (not yesterday) so a caller polling for "today's spend so far" gets the partial row rather than silence — the row is flagged ``is_provisional``, which is the honest answer. """ end_dt = end or datetime.now(UTC).date() start_dt = start or (end_dt - timedelta(days=29)) if start_dt > end_dt: raise ValidationError("`start` must be on or before `end`.") span = (end_dt - start_dt).days + 1 if span > max_days: raise ValidationError( f"Range too large: {span} days requested, max is {max_days}. " + narrow_hint, ) return start_dt, end_dt def _parse_since(raw: str) -> datetime: """Parse an incremental cursor leniently. Three real-world shapes all mean the same instant and all arrive: * ``2026-08-14T02:11:07+00:00`` — correct, and correctly encoded * ``2026-08-14T02:11:07 00:00`` — the same value with the ``+`` eaten by form decoding, which is what happens when a caller pastes a timestamp into a URL without escaping it. Rejecting it teaches the integrator nothing; the intent is unambiguous. * ``2026-08-14T02:11:07Z`` / no offset at all — Zulu or naive, read as UTC (every timestamp this API emits is UTC). """ value = (raw or "").strip() if not value: raise ValidationError("`since` is required (ISO-8601 timestamp).") candidates = [value] if value.endswith("Z"): candidates.append(value[:-1] + "+00:00") if " " in value: head, _, tail = value.rpartition(" ") candidates.append(f"{head}+{tail}") for candidate in candidates: try: return datetime.fromisoformat(candidate) except ValueError: continue raise ValidationError( f"`since` must be an ISO-8601 timestamp (got {raw!r}), e.g. " "2026-08-14T02:11:07Z. Remember to URL-encode a `+` offset.", ) def _resolve_incremental( since: datetime | str, *, horizon_days: int = DEFAULT_HORIZON_DAYS, ) -> tuple[datetime, date]: """Normalize ``since`` to UTC and derive the detection horizon. A naive timestamp is read as UTC rather than rejected: every other timestamp this API emits is UTC, and 422-ing a caller who echoed one back without its offset would be pedantry. A ``since`` in the future is refused — it would return an empty page that reads as "nothing changed" when the truth is "your clock is wrong". """ parsed = _parse_since(since) if isinstance(since, str) else since normalized = ( parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) ).astimezone(UTC) now = datetime.now(UTC) if normalized > now + timedelta(minutes=5): raise ValidationError( "`since` is in the future; an empty result would read as " "'nothing changed'. Check the caller's clock.", ) horizon = horizon_start(horizon_days, today=now.date()) if normalized.date() < horizon: raise ValidationError( f"`since` ({normalized.date()}) predates the detection horizon " f"({horizon}; horizon_days max {MAX_HORIZON_DAYS}). Changes older " "than the horizon are not tracked — do a full pull for that " "window instead of an incremental one.", ) return normalized, horizon def _parse_month(value: str, *, field: str) -> date: try: year_str, month_str = value.strip().split("-") parsed = date(int(year_str), int(month_str), 1) except (ValueError, AttributeError) as exc: raise ValidationError( f"`{field}` must be a month in YYYY-MM form (got {value!r}).", ) from exc return parsed def _month_span_months(first: date, last: date) -> int: return (last.year - first.year) * 12 + (last.month - first.month) + 1 def _normalize_group_by(raw: str) -> str: """Lower-case dimension names while preserving tag-key case. ``tag:UsedBy`` and ``tag:usedby`` are different keys in the cloud's own data (GCP labels are lower-case by rule, AWS tags are not), so folding the whole string would turn a valid request into one that matches nothing — and an empty result reads as "this team spent nothing", which is the worst possible failure for a cost API. """ parts = [] for part in (raw or "").split(","): piece = part.strip() if piece.lower().startswith(TAG_PREFIX): parts.append(TAG_PREFIX + piece[len(TAG_PREFIX):].strip()) else: parts.append(piece.lower()) return ",".join(parts) def _validate_environments(values: list[str] | None) -> tuple[str, ...]: if not values: return () unknown = sorted({v for v in values if v not in _VALID_ENVIRONMENTS}) if unknown: raise ValidationError( f"environment must be one of {sorted(_VALID_ENVIRONMENTS)}; " f"got {', '.join(unknown)}", ) return tuple(dict.fromkeys(values)) async def _audit_read( db: AsyncSession, request: Request, ctx: ApiClientContext, *, action: str, query: BillingQuery, row_count: int, extra: dict | None = None, ) -> None: """One row per successful billing read. Cloud spend for the whole company is the kind of data whose export must be reconstructable months later ("who pulled a year of every account, and when"). Attributed to the service principal — no human actor is implied for a machine call. Never raises: an audit failure must not deny a read that already succeeded (it is logged instead). "Not raising" is not enough on its own, though — a failed flush deactivates the whole transaction, so a merely-swallowed error leaves the session unusable and ``get_db``'s closing commit turns an already-served read into a 500. Hence ``best_effort``: the write runs inside a SAVEPOINT, and a failure rolls back only that. """ from app.core.txn import best_effort # SAVEPOINT-isolated rather than a bare try/except: a failed flush # deactivates the WHOLE transaction, so merely swallowing the error # leaves the session unusable and ``get_db``'s closing commit turns an # already-served read into a 500. This is the repo's primitive for # exactly that bug class (see app/core/txn.py). async with best_effort(db, label="cloud-billing read audit") as step: from app.audit.service import AuditService await AuditService(db).log( trace_id=( getattr(request.state, "trace_id", None) or str(uuid_mod.uuid4()) ), actor_type="service", action=action, status="success", resource_type="cloud_billing", resource_id=ctx.slug, detail={ "client_slug": ctx.slug, "window": { "start": query.start.isoformat(), "end": query.end.isoformat(), }, "subscriptions": list(query.subscriptions) or None, "providers": list(query.providers) or None, "environments": list(query.environments) or None, "owner_email": query.owner_email, "include_services": query.include_services, "rows": row_count, # Stamped so a reader of this row never has to guess # which cost basis left the building. "measure": "list", **(extra or {}), }, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) if step.ok: await db.commit() else: logger.warning( "cloud-billing read audit failed (client=%s): %s", ctx.slug, step.detail, ) def _page( result: BillingResult, *, page: int, page_size: int, ) -> tuple[list[dict], int]: total = len(result.rows) offset = (page - 1) * page_size return result.rows[offset:offset + page_size], total # ── Endpoints ───────────────────────────────────────────────────────── @router.get( "/daily", summary="Daily cloud spend by subscription", responses=_example(_DAILY_EXAMPLE), ) async def daily_billing( request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_CLOUD_BILLING_READ)), db: AsyncSession = Depends(get_db), start: date | None = Query( default=None, description="Inclusive start date; defaults to 29 days ago.", ), end: date | None = Query( default=None, description="Inclusive end date; defaults to today (UTC).", ), subscription: list[str] | None = Query( default=None, description=( "Repeatable. Helix subscription id (UUID) or cloud-native " "external id. Outside your visibility → matches nothing." ), ), provider: list[str] | None = Query(default=None, description="Repeatable."), environment: list[str] | None = Query( default=None, description="Repeatable: dev / staging / prod.", ), owner_email: str | None = Query(default=None, max_length=255), group_by: str = Query( default="subscription", description=( "`subscription` (default) or `subscription,service` — the " "latter adds a `service` field and caps the window at 92 days." ), ), updated_since: str | None = Query( default=None, description=( "Incremental pull: return only (subscription, day) cells " "re-ingested at or after this timestamp. See /billing/changes " "for the exact semantics and the horizon limit." ), ), horizon_days: int = Query( default=DEFAULT_HORIZON_DAYS, ge=1, le=MAX_HORIZON_DAYS, description=( "Change-detection look-back for `updated_since` — same knob " "and same limits as on /billing/changes. Ignored without " "`updated_since`." ), ), page: int = Query(default=1, ge=1), page_size: int = Query(default=_DEFAULT_PAGE_SIZE, ge=1, le=_MAX_PAGE_SIZE), ) -> dict: """One row per (day × subscription), optionally × service.""" group_by_clean = group_by.replace(" ", "") if group_by_clean not in _VALID_DAILY_GROUP_BY: raise ValidationError( f"group_by must be one of {sorted(_VALID_DAILY_GROUP_BY)}", ) include_services = group_by_clean == "subscription,service" start_dt, end_dt = _resolve_day_window( start, end, max_days=_MAX_SERVICE_SPAN_DAYS if include_services else _MAX_DAILY_SPAN_DAYS, narrow_hint=( "Narrow the window, or drop `group_by=subscription,service` " "(the service breakdown has a tighter limit)." if include_services else "Narrow the window." ), ) query = BillingQuery( start=start_dt, end=end_dt, subscriptions=tuple(subscription or ()), providers=tuple(provider or ()), environments=_validate_environments(environment), owner_email=owner_email, include_services=include_services, ) visibility = await resolve_client_cloud_visibility(db, ctx) subs = await resolve_subscriptions(db, visibility=visibility, query=query) result = await daily_costs(db, subs=subs, query=query) incremental_meta: dict = {} if updated_since is not None: since, horizon = _resolve_incremental( updated_since, horizon_days=horizon_days, ) cells, signals = await changed_cells( db, subs=subs, since=since, horizon=horizon, ) # Filter BEFORE paging: a page computed over unfiltered rows would # be mostly empty and the total would be a different question's # answer. result.rows = [ row for row in result.rows if (row["subscription_id"], date.fromisoformat(row["date"])) in cells ] incremental_meta = { "incremental": changes_meta( since=since, horizon=horizon, signals=signals, cell_count=len(cells), ), } rows, total = _page(result, page=page, page_size=page_size) await _audit_read( db, request, ctx, action="cloud.billing.read_daily", query=query, row_count=total, extra={"updated_since": updated_since}, ) return success_response( data=rows, request=request, meta={ "page": page, "page_size": page_size, "total": total, **response_meta(result, query=query), **incremental_meta, }, ) @router.get( "/monthly", summary="Monthly cloud spend by subscription", responses=_example(_MONTHLY_EXAMPLE), ) async def monthly_billing( request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_CLOUD_BILLING_READ)), db: AsyncSession = Depends(get_db), start_month: str | None = Query( default=None, description="Inclusive, `YYYY-MM`; defaults to 11 months ago.", ), end_month: str | None = Query( default=None, description="Inclusive, `YYYY-MM`; defaults to this month.", ), subscription: list[str] | None = Query(default=None), provider: list[str] | None = Query(default=None), environment: list[str] | None = Query(default=None), owner_email: str | None = Query(default=None, max_length=255), group_by: str = Query(default="subscription"), page: int = Query(default=1, ge=1), page_size: int = Query(default=_DEFAULT_PAGE_SIZE, ge=1, le=_MAX_PAGE_SIZE), ) -> dict: """One row per (month × subscription), optionally × service. Folded from the same daily cells the `/daily` endpoint returns, so a month total always equals the sum of its days on this API. """ group_by_clean = group_by.replace(" ", "") if group_by_clean not in _VALID_DAILY_GROUP_BY: raise ValidationError( f"group_by must be one of {sorted(_VALID_DAILY_GROUP_BY)}", ) include_services = group_by_clean == "subscription,service" today = datetime.now(UTC).date() last = _parse_month(end_month, field="end_month") if end_month else date( today.year, today.month, 1, ) if start_month: first = _parse_month(start_month, field="start_month") else: # 12 months inclusive of the end month. month_index = last.year * 12 + (last.month - 1) - 11 first = date(month_index // 12, month_index % 12 + 1, 1) if first > last: raise ValidationError("`start_month` must be on or before `end_month`.") span = _month_span_months(first, last) max_months = 3 if include_services else _MAX_MONTHLY_SPAN_MONTHS if span > max_months: raise ValidationError( f"Range too large: {span} months requested, max is {max_months}" + (" with the service breakdown." if include_services else "."), ) # Inclusive end = last day of the end month, clamped to today so a # window running into the future never reports a "complete" month. if last.month == 12: month_end = date(last.year, 12, 31) else: month_end = date(last.year, last.month + 1, 1) - timedelta(days=1) query = BillingQuery( start=first, end=min(month_end, today), subscriptions=tuple(subscription or ()), providers=tuple(provider or ()), environments=_validate_environments(environment), owner_email=owner_email, include_services=include_services, ) visibility = await resolve_client_cloud_visibility(db, ctx) subs = await resolve_subscriptions(db, visibility=visibility, query=query) result = await monthly_costs(db, subs=subs, query=query) rows, total = _page(result, page=page, page_size=page_size) await _audit_read( db, request, ctx, action="cloud.billing.read_monthly", query=query, row_count=total, ) meta = response_meta(result, query=query) meta["months"] = {"start": f"{first.year:04d}-{first.month:02d}", "end": f"{last.year:04d}-{last.month:02d}"} return success_response( data=rows, request=request, meta={"page": page, "page_size": page_size, "total": total, **meta}, ) @router.get( "/breakdown", summary="Cloud spend for a window, grouped by one dimension", responses=_example(_BREAKDOWN_EXAMPLE), ) async def billing_breakdown( request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_CLOUD_BILLING_READ)), db: AsyncSession = Depends(get_db), start: date | None = Query(default=None), end: date | None = Query(default=None), group_by: str = Query( default="subscription", description=( "One or two comma-separated dimensions: subscription / service " "/ provider / environment / department / owner / resource / " "region / resource_type / cost_center, or `tag:` for a " "cloud resource tag (see GET /billing/tag-keys). Crossing two " "answers questions like `tag:team,environment`." ), ), top: int = Query( default=DEFAULT_TOP, ge=1, le=MAX_TOP, description=( "Row cap (rows are ordered by spend, largest first). " "`meta.truncated_to_top` says when it applied." ), ), subscription: list[str] | None = Query(default=None), provider: list[str] | None = Query(default=None), environment: list[str] | None = Query(default=None), owner_email: str | None = Query(default=None, max_length=255), ) -> dict: """Window total split along one dimension, largest first. Window cap is 92 days — deliberately tighter than `/daily`'s 366: a grouping aggregates every line item in the window in one pass, and the dimensions with real cardinality (service, resource) are exactly the ones a year-wide request would be most tempting for. """ # Tag keys are case-sensitive as the cloud stores them, so only the # dimension NAMES are lower-cased — ``tag:UsedBy`` must not silently # become ``tag:usedby`` and match nothing. dims = parse_dimensions(_normalize_group_by(group_by)) group = ",".join(d.spec for d in dims) start_dt, end_dt = _resolve_day_window( start, end, max_days=_MAX_BREAKDOWN_SPAN_DAYS, narrow_hint=( "Narrow the window, or filter to fewer subscriptions " "(`subscription` / `provider` / `environment` / `owner_email`)." ), ) query = BillingQuery( start=start_dt, end=end_dt, subscriptions=tuple(subscription or ()), providers=tuple(provider or ()), environments=_validate_environments(environment), owner_email=owner_email, ) visibility = await resolve_client_cloud_visibility(db, ctx) subs = await resolve_subscriptions(db, visibility=visibility, query=query) result = await breakdown(db, subs=subs, query=query, dims=dims, top=top) await _audit_read( db, request, ctx, action="cloud.billing.read_breakdown", query=query, row_count=len(result.rows), extra={"group_by": group}, ) meta = { "measure": "list", "measure_note": ( "Business pre-discount (list) cost. Not the invoiced net " "amount — do not reconcile against a provider invoice." ), "currency": "USD", "window": {"start": start_dt.isoformat(), "end": end_dt.isoformat()}, "excludes_unallocatable": True, **breakdown_meta(result, group_by=group), } return success_response(data=result.rows, request=request, meta=meta) @router.get( "/changes", summary="Which (subscription, day) cells changed since a timestamp", responses=_example(_CHANGES_EXAMPLE), ) async def billing_changes( request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_CLOUD_BILLING_READ)), db: AsyncSession = Depends(get_db), since: str = Query( description=( "ISO-8601 timestamp. Naive values are read as UTC; a `Z` " "suffix and an unescaped `+` offset are both accepted." ), ), horizon_days: int = Query( default=DEFAULT_HORIZON_DAYS, ge=1, le=MAX_HORIZON_DAYS, description=( "How far back to look for changes. Changes to days older than " "this are not detected — re-pull such a window explicitly." ), ), subscription: list[str] | None = Query(default=None), provider: list[str] | None = Query(default=None), environment: list[str] | None = Query(default=None), owner_email: str | None = Query(default=None, max_length=255), page: int = Query(default=1, ge=1), page_size: int = Query(default=_DEFAULT_PAGE_SIZE, ge=1, le=_MAX_PAGE_SIZE), ) -> dict: """The incremental-pull driver: re-pull only these days. An *ingest watermark*, not an amount diff — a day re-fetched with an unchanged amount is still listed, because a consumer that overwrites by primary key does not care and under-reporting would be a silent data loss on their side. """ since_dt, horizon = _resolve_incremental(since, horizon_days=horizon_days) query = BillingQuery( start=horizon, end=datetime.now(UTC).date(), subscriptions=tuple(subscription or ()), providers=tuple(provider or ()), environments=_validate_environments(environment), owner_email=owner_email, ) visibility = await resolve_client_cloud_visibility(db, ctx) subs = await resolve_subscriptions(db, visibility=visibility, query=query) cells, signals = await changed_cells( db, subs=subs, since=since_dt, horizon=horizon, ) rows = changes_rows(cells, subs=subs) total = len(rows) offset = (page - 1) * page_size await _audit_read( db, request, ctx, action="cloud.billing.read_changes", query=query, row_count=total, extra={"since": since_dt.isoformat(), "horizon_days": horizon_days}, ) return success_response( data=rows[offset:offset + page_size], request=request, meta={ "page": page, "page_size": page_size, "total": total, **changes_meta( since=since_dt, horizon=horizon, signals=signals, cell_count=len(cells), ), }, ) @router.get( "/tag-keys", summary="Which cloud resource tags exist on your visible spend", responses=_example(_TAG_KEYS_EXAMPLE), ) async def billing_tag_keys( request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_CLOUD_BILLING_READ)), db: AsyncSession = Depends(get_db), start: date | None = Query(default=None), end: date | None = Query(default=None), subscription: list[str] | None = Query(default=None), provider: list[str] | None = Query(default=None), environment: list[str] | None = Query(default=None), owner_email: str | None = Query(default=None, max_length=255), limit: int = Query(default=100, ge=1, le=500), ) -> dict: """Discovery for `group_by=tag:` — you cannot slice by a tag you do not know exists. Ranked by covered spend, so the keys that carry real money sort above the ones a Kubernetes controller sprays onto every node. Values are NOT returned: ask for them with `group_by=tag:`, which applies the same visibility rules as everything else here. """ start_dt, end_dt = _resolve_day_window( start, end, max_days=_MAX_BREAKDOWN_SPAN_DAYS, narrow_hint="Narrow the window, or filter to fewer subscriptions.", ) query = BillingQuery( start=start_dt, end=end_dt, subscriptions=tuple(subscription or ()), providers=tuple(provider or ()), environments=_validate_environments(environment), owner_email=owner_email, ) visibility = await resolve_client_cloud_visibility(db, ctx) subs = await resolve_subscriptions(db, visibility=visibility, query=query) rows, sampled = await tag_keys(db, subs=subs, query=query, limit=limit) await _audit_read( db, request, ctx, action="cloud.billing.read_tag_keys", query=query, row_count=len(rows), ) return success_response( data=rows, request=request, meta={ "window": {"start": start_dt.isoformat(), "end": end_dt.isoformat()}, "sampled_rows": sampled, "sample_limit_per_month": TAG_KEY_SCAN_LIMIT, # Said plainly because the number looks like a total and is # not one: `tags` is an unindexed JSON column with no # JSON_KEYS pushdown, so keys come off a bounded, recency- # biased sample. Rank by it; never quote it as spend. "sampled_spend_note": ( "sampled_spend_usd ranks keys within a bounded sample — " "it is not the key's total spend. Use " "/billing/breakdown?group_by=tag: for exact amounts." ), }, ) __all__ = ["router"]