CUR vs Cost Explorer API for Programmatic Cost Data

"""Facade over two AWS cost-data backends: CUR-via-Athena and the Cost Explorer API.

Callers depend on one CostDataProvider-style interface; the facade picks the
backend based on the query shape so pipelines never hardcode which AWS API
their cost figures came from.
"""
from __future__ import annotations

import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import date, timedelta
from typing import Iterator, Optional

import boto3
from botocore.exceptions import ClientError

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

ATHENA_DATABASE = "finops_billing"
ATHENA_TABLE = "cur_parquet"
ATHENA_OUTPUT_S3 = "s3://finops-athena-results/cur-vs-ce/"
CE_REQUEST_COST_USD = 0.01  # billed per paginated GetCostAndUsage call beyond the free quota


def with_retries(max_attempts: int = 5, base_delay: float = 1.5):
    """Exponential backoff shared by the Athena and Cost Explorer clients."""
    def decorator(fn):
        def wrapped(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except ClientError as exc:
                    code = exc.response.get("Error", {}).get("Code", "")
                    retryable = code in ("ThrottlingException", "TooManyRequestsException")
                    if not retryable or attempt == max_attempts:
                        raise
                    delay = base_delay * (2 ** (attempt - 1))
                    logger.warning("Throttled (%s), retry %d/%d in %.1fs", code, attempt, max_attempts, delay)
                    time.sleep(delay)
        return wrapped
    return decorator


@dataclass
class CostQuery:
    start_date: date
    end_date: date
    granularity: str = "DAILY"                       # DAILY | MONTHLY | HOURLY
    group_by: list[str] = field(default_factory=lambda: ["SERVICE"])
    resource_level: bool = False                      # True forces the CUR/Athena backend


@dataclass
class CostRecord:
    period_start: str
    dimension_key: str
    amount: float
    currency: str
    resource_id: Optional[str] = None
    source: str = ""


class CostDataBackend(ABC):
    """One programmatic-cost-data interface, two AWS implementations."""

    @abstractmethod
    def fetch(self, query: CostQuery) -> Iterator[CostRecord]:
        ...


class AthenaCURBackend(CostDataBackend):
    """Resource-level, hourly-capable reads against the CUR Parquet lake."""

    def __init__(self, athena_client=None):
        self.client = athena_client or boto3.client("athena")

    @with_retries()
    def _start_query(self, sql: str) -> str:
        resp = self.client.start_query_execution(
            QueryString=sql,
            QueryExecutionContext={"Database": ATHENA_DATABASE},
            ResultConfiguration={"OutputLocation": ATHENA_OUTPUT_S3},
        )
        return resp["QueryExecutionId"]

    def _wait_for_completion(self, execution_id: str, poll_seconds: float = 2.0) -> None:
        while True:
            status = self.client.get_query_execution(QueryExecutionId=execution_id)
            state = status["QueryExecution"]["Status"]["State"]
            if state == "SUCCEEDED":
                return
            if state in ("FAILED", "CANCELLED"):
                raise RuntimeError(f"Athena query {execution_id} ended in {state}")
            time.sleep(poll_seconds)

    def fetch(self, query: CostQuery) -> Iterator[CostRecord]:
        resource_col = ", line_item_resource_id" if query.resource_level else ""
        sql = f"""
        SELECT line_item_usage_start_date, product_servicecode,
               SUM(line_item_unblended_cost) AS cost, line_item_currency_code{resource_col}
        FROM {ATHENA_TABLE}
        WHERE line_item_usage_start_date BETWEEN DATE '{query.start_date}' AND DATE '{query.end_date}'
        GROUP BY line_item_usage_start_date, product_servicecode, line_item_currency_code{resource_col}
        """
        execution_id = self._start_query(sql)
        self._wait_for_completion(execution_id)
        paginator = self.client.get_paginator("get_query_results")
        for page in paginator.paginate(QueryExecutionId=execution_id):
            for row in page["ResultSet"]["Rows"][1:]:  # skip header row
                cells = [c.get("VarCharValue", "") for c in row["Data"]]
                yield CostRecord(
                    period_start=cells[0], dimension_key=cells[1], amount=float(cells[2] or 0),
                    currency=cells[3], resource_id=cells[4] if query.resource_level else None,
                    source="cur_athena",
                )


class CostExplorerBackend(CostDataBackend):
    """Aggregated, near-real-time reads against GetCostAndUsage."""

    def __init__(self, ce_client=None):
        self.client = ce_client or boto3.client("ce")
        self.billed_requests = 0

    @with_retries()
    def _page(self, query: CostQuery, next_token: Optional[str]):
        kwargs = dict(
            TimePeriod={"Start": str(query.start_date), "End": str(query.end_date)},
            Granularity=query.granularity,
            Metrics=["UnblendedCost"],
            GroupBy=[{"Type": "DIMENSION", "Key": k} for k in query.group_by],
        )
        if next_token:
            kwargs["NextPageToken"] = next_token
        self.billed_requests += 1  # every page, including the first, is a billed request
        return self.client.get_cost_and_usage(**kwargs)

    def fetch(self, query: CostQuery) -> Iterator[CostRecord]:
        next_token = None
        while True:
            resp = self._page(query, next_token)
            for result in resp["ResultsByTime"]:
                for group in result["Groups"]:
                    metric = group["Metrics"]["UnblendedCost"]
                    yield CostRecord(
                        period_start=result["TimePeriod"]["Start"], dimension_key="|".join(group["Keys"]),
                        amount=float(metric["Amount"]), currency=metric["Unit"], source="cost_explorer",
                    )
            next_token = resp.get("NextPageToken")
            if not next_token:
                break
        logger.info("Cost Explorer billed %d requests (~$%.2f)", self.billed_requests,
                    self.billed_requests * CE_REQUEST_COST_USD)


class CostDataFacade:
    """Single entry point; routes each query to the backend that can satisfy it."""

    def __init__(self):
        self._cur = AthenaCURBackend()
        self._ce = CostExplorerBackend()

    def fetch(self, query: CostQuery) -> Iterator[CostRecord]:
        # Resource identity and hourly granularity exist only in CUR.
        # Everything else prefers Cost Explorer for lower setup cost and faster availability.
        if query.resource_level or query.granularity == "HOURLY":
            logger.info("Routing to CUR/Athena (resource_level=%s, granularity=%s)",
                        query.resource_level, query.granularity)
            return self._cur.fetch(query)
        logger.info("Routing to Cost Explorer (granularity=%s)", query.granularity)
        return self._ce.fetch(query)


if __name__ == "__main__":
    facade = CostDataFacade()

    # Daily service totals for a dashboard refresh -> Cost Explorer.
    dashboard_query = CostQuery(start_date=date.today() - timedelta(days=7), end_date=date.today())
    for record in facade.fetch(dashboard_query):
        logger.info("%s | %s | $%.2f", record.source, record.dimension_key, record.amount)

    # Resource-level audit for a closed month -> CUR/Athena.
    audit_query = CostQuery(start_date=date(2026, 6, 1), end_date=date(2026, 6, 30), resource_level=True)
    for record in facade.fetch(audit_query):
        logger.info("%s | %s (%s) | $%.2f", record.source, record.dimension_key, record.resource_id, record.amount)