capella_billing_per_support_plan.py

#!/usr/bin/env python3

"""
Capella Billing Per Support Plan Script
--------------------------------------------------

Description:
    This script helps identify all clusters belonging to a
    specific support plan within a Capella Organization and
    fetches aggregated billing information for those clusters.

Flow:
    1. Fetch all projects in an organization
    2. Fetch all clusters across projects
    3. Identify clusters by support plan
    4. Fetch aggregated org billing using filtered instanceIds
    5. Optionally return only the billing total section

Supported Support Plans:
    - developer pro
    - basic
    - enterprise

Authentication:
    Update the bearer token directly in CONFIG.

Dependencies:
    pip install requests

Usage Examples:

    Full billing response:
    ------------------------------------------------
    python3 capella_billing_per_support_plan.py \
      --org-id "<ORG_ID>" \
      --start-date "2026-04-01" \
      --end-date "2026-04-30" \
      --support-plan "enterprise"

    Total only response:
    ------------------------------------------------
    python3 capella_billing_per_support_plan.py \
      --org-id "<ORG_ID>" \
      --start-date "2026-04-01" \
      --end-date "2026-04-30" \
      --support-plan "enterprise" \
      --total-only true

    Show all available parameters:
    ------------------------------------------------
    python3 capella_billing_per_support_plan.py --help

Parameters:
    --org-id
        Capella Organization ID

    --start-date
        Billing start date in YYYY-MM-DD format. Maximum supported date range is 366 days

    --end-date
        Billing end date in YYYY-MM-DD format. Maximum supported date range is 366 days

    --support-plan
        Supported values:
            developer pro
            basic
            enterprise

    --total-only
        Optional.
        If set to true, only the "total" section
        of the billing response is printed.

        Supported values:
            true
            false

        Default:
            false
"""

import argparse
import json
import sys
import time
from dataclasses import dataclass
from typing import Dict, List

import requests

CONFIG = {
    "bearer_token": "<CAPELLA_BEARER_TOKEN>",
    "base_url": "https://cloudapi.cloud.couchbase.com"
}

SUPPORTED_SUPPORT_PLANS = {
    "developer pro",
    "basic",
    "enterprise"
}

# ============================================================
# RATE LIMITING
# ============================================================

RATE_LIMIT_PER_MINUTE = 90
RATE_LIMIT_WINDOW_SECONDS = 60

request_counter = 0
window_start_time = time.time()


@dataclass
class Project:
    project_id: str
    project_name: str


@dataclass
class Cluster:
    cluster_id: str
    cluster_name: str
    project_id: str
    project_name: str


@dataclass
class ClusterBilling:
    cluster_id: str
    cluster_name: str
    project_id: str
    project_name: str
    support_plan: str


def str_to_bool(value):

    return str(value).strip().lower() == "true"




def apply_rate_limit():

    global request_counter
    global window_start_time

    current_time = time.time()

    elapsed = current_time - window_start_time

    if elapsed >= RATE_LIMIT_WINDOW_SECONDS:

        request_counter = 0
        window_start_time = current_time

    if request_counter >= RATE_LIMIT_PER_MINUTE:

        sleep_time = (
            RATE_LIMIT_WINDOW_SECONDS - elapsed
        )

        if sleep_time > 0:

            print("\n===================================================")
            print("RATE LIMIT PROTECTION")
            print("===================================================")

            print(
                f"\nApproaching Capella Management API "
                f"rate limit."
            )

            print(
                f"Sleeping for "
                f"{round(sleep_time, 2)} seconds..."
            )

            time.sleep(sleep_time)

        request_counter = 0
        window_start_time = time.time()

    request_counter += 1


def create_session() -> requests.Session:

    if not CONFIG["bearer_token"]:

        print("Bearer token not configured.")
        sys.exit(1)

    session = requests.Session()

    session.headers.update({
        "Authorization": f"Bearer {CONFIG['bearer_token']}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    })

    return session


def handle_response(response: requests.Response) -> Dict:

    if response.status_code >= 400:

        print("\nERROR:")
        print(f"Status Code : {response.status_code}")
        print(f"Response    : {response.text}")

        sys.exit(1)

    return response.json()


# ============================================================
# PROJECTS
# ============================================================

def get_projects(session, org_id):

    projects = []

    url = (
        f"{CONFIG['base_url']}"
        f"/v4/organizations/{org_id}/projects"
    )

    print("\n===================================================")
    print("STEP 1 - FETCHING PROJECTS")
    print("===================================================")

    while url:

        print(f"\nGET {url}")

        apply_rate_limit()
        response = session.get(url)

        data = handle_response(response)

        for item in data.get("data", []):

            projects.append(
                Project(
                    project_id=item["id"],
                    project_name=item["name"]
                )
            )

        next_page = (
            data.get("cursor", {})
                .get("pages", {})
                .get("next")
        )

        if next_page:

            url = (
                f"{CONFIG['base_url']}"
                f"/v4/organizations/{org_id}"
                f"/projects?page={next_page}"
            )

        else:
            url = None

    print(f"\nTotal Projects Found: {len(projects)}")

    return projects


# ============================================================
# CLUSTERS
# ============================================================

def get_clusters(session, org_id, projects):

    all_clusters = []

    print("\n===================================================")
    print("STEP 2 - FETCHING CLUSTERS")
    print("===================================================")

    for project in projects:

        print(
            f"\nFetching clusters for project: "
            f"{project.project_name}"
        )

        url = (
            f"{CONFIG['base_url']}"
            f"/v4/organizations/{org_id}"
            f"/projects/{project.project_id}/clusters"
        )

        while url:

            print(f"GET {url}")

            apply_rate_limit()
            response = session.get(url)

            data = handle_response(response)

            for item in data.get("data", []):

                cluster = Cluster(
                    cluster_id=item["id"],
                    cluster_name=item["name"],
                    project_id=project.project_id,
                    project_name=project.project_name
                )

                all_clusters.append(cluster)

                print(
                    f"  - {cluster.cluster_name} "
                    f"({cluster.cluster_id})"
                )

            next_page = (
                data.get("cursor", {})
                    .get("pages", {})
                    .get("next")
            )

            if next_page:

                url = (
                    f"{CONFIG['base_url']}"
                    f"/v4/organizations/{org_id}"
                    f"/projects/{project.project_id}"
                    f"/clusters?page={next_page}"
                )

            else:
                url = None

    print(f"\nTOTAL CLUSTERS FOUND: {len(all_clusters)}")

    return all_clusters


# ============================================================
# SUPPORT PLAN LOOKUP
# ============================================================

def find_key_recursively(obj, target_key):

    if isinstance(obj, dict):

        for key, value in obj.items():

            if key == target_key:
                return value

            result = find_key_recursively(value, target_key)

            if result:
                return result

    elif isinstance(obj, list):

        for item in obj:

            result = find_key_recursively(item, target_key)

            if result:
                return result

    return None


def extract_support_plan(data):

    found = find_key_recursively(data, "supportPlan")

    if found:
        return str(found)

    return None


def get_cluster_support_plan(
    session,
    org_id,
    cluster,
    start_date,
    end_date
):

    url = (
        f"{CONFIG['base_url']}"
        f"/v4/organizations/{org_id}"
        f"/projects/{cluster.project_id}"
        f"/clusters/{cluster.cluster_id}"
        f"/billing"
    )

    payload = {
        "startDate": start_date,
        "endDate": end_date
    }

    print(f"\nPOST {url}")

    apply_rate_limit()
    response = session.post(
        url,
        json=payload
    )

    if response.status_code >= 400:

        print("\n===================================================")
        print("BILLING LOOKUP FAILED")
        print("===================================================")

        print(
            f"\nCluster Name : "
            f"{cluster.cluster_name}"
        )

        print(
            f"Project Name : "
            f"{cluster.project_name}"
        )

        print(
            f"Project ID   : "
            f"{cluster.project_id}"
        )

        print(
            f"Cluster ID   : "
            f"{cluster.cluster_id}"
        )

        print(
            f"Status Code  : "
            f"{response.status_code}"
        )

        print(
            f"Response     : "
            f"{response.text}"
        )

        sys.exit(1)

    data = response.json()

    return extract_support_plan(data)


def filter_clusters_by_support_plan(
    session,
    org_id,
    clusters,
    start_date,
    end_date,
    requested_support_plan
):

    matching_clusters = []

    print("\n===================================================")
    print("STEP 3 - FILTERING CLUSTERS BY SUPPORT PLAN")
    print("===================================================")

    for cluster in clusters:

        print(
            f"\nChecking cluster: "
            f"{cluster.cluster_name}"
        )

        support_plan = get_cluster_support_plan(
            session=session,
            org_id=org_id,
            cluster=cluster,
            start_date=start_date,
            end_date=end_date
        )

        if not support_plan:

            print("Support Plan: NOT FOUND")
            continue

        print(f"Support Plan: {support_plan}")

        if (
            support_plan.lower()
            == requested_support_plan.lower()
        ):

            print("MATCH FOUND")

            matching_clusters.append(
                ClusterBilling(
                    cluster_id=cluster.cluster_id,
                    cluster_name=cluster.cluster_name,
                    project_id=cluster.project_id,
                    project_name=cluster.project_name,
                    support_plan=support_plan
                )
            )

    print(
        f"\nTOTAL MATCHING CLUSTERS: "
        f"{len(matching_clusters)}"
    )

    return matching_clusters


# ============================================================
# BILLING
# ============================================================

def print_total_only(data):

    total = (
        data.get("total")
        or data.get("data", {}).get("total")
    )

    print("\n===================================================")
    print("TOTAL ONLY RESPONSE")
    print("===================================================")

    if not total:

        print("\nNo total section found in response.")
        return

    print(json.dumps(total, indent=2))


def get_aggregated_org_billing(
    session,
    org_id,
    matching_clusters,
    start_date,
    end_date,
    total_only
):

    print("\n===================================================")
    print("STEP 4 - FETCHING AGGREGATED BILLING")
    print("===================================================")

    instance_ids = [
        cluster.cluster_id
        for cluster in matching_clusters
    ]

    print("\nCluster IDs used in billing filter:")

    for cluster_id in instance_ids:
        print(f"  - {cluster_id}")

    url = (
        f"{CONFIG['base_url']}"
        f"/v4/organizations/{org_id}"
        f"/billing"
    )

    payload = {
        "startDate": start_date,
        "endDate": end_date,
        "filters": {
            "instanceIds": instance_ids
        }
    }

    print("\nCalling Aggregated Billing API:")
    print(f"POST {url}")

    print("\nRequest Payload:")
    print(json.dumps(payload, indent=2))

    apply_rate_limit()
    response = session.post(
        url,
        json=payload
    )

    data = handle_response(response)

    if total_only:

        print_total_only(data)

    else:

        print("\n===================================================")
        print("FULL BILLING RESPONSE")
        print("===================================================")

        print(json.dumps(data, indent=2))

    return data


# ============================================================
# MAIN
# ============================================================

def main():

    parser = argparse.ArgumentParser(
        description="Capella Billing Aggregator"
    )

    parser.add_argument(
        "--org-id",
        required=True,
        help="Capella Organization ID"
    )

    parser.add_argument(
        "--start-date",
        required=True,
        help="Billing start date in YYYY-MM-DD format. Maximum supported date range is 366 days"
    )

    parser.add_argument(
        "--end-date",
        required=True,
        help="Billing end date in YYYY-MM-DD format. Maximum supported date range is 366 days"
    )

    parser.add_argument(
        "--support-plan",
        required=True,
        help=(
            "Supported values: "
            "developer pro, basic, enterprise"
        )
    )

    parser.add_argument(
        "--total-only",
        default="false",
        help=(
            "If true, only billing total section "
            "is printed. Supported values: true/false"
        )
    )

    args = parser.parse_args()

    normalized_support_plan = (
        args.support_plan.strip().lower()
    )

    if normalized_support_plan not in SUPPORTED_SUPPORT_PLANS:

        print("\nERROR:")
        print("Unsupported support plan provided.")

        print("\nSupported values are:")

        for value in sorted(SUPPORTED_SUPPORT_PLANS):
            print(f"  - {value}")

        sys.exit(1)

    total_only = str_to_bool(args.total_only)

    print("\n===================================================")
    print("CAPELLA BILLING AGGREGATION SCRIPT")
    print("===================================================")

    print(f"\nOrganization ID : {args.org_id}")
    print(f"Start Date      : {args.start_date}")
    print(f"End Date        : {args.end_date}")
    print(f"Support Plan    : {normalized_support_plan}")
    print(f"Total Only      : {total_only}")

    session = create_session()

    projects = get_projects(
        session=session,
        org_id=args.org_id
    )

    clusters = get_clusters(
        session=session,
        org_id=args.org_id,
        projects=projects
    )

    matching_clusters = (
        filter_clusters_by_support_plan(
            session=session,
            org_id=args.org_id,
            clusters=clusters,
            start_date=args.start_date,
            end_date=args.end_date,
            requested_support_plan=normalized_support_plan
        )
    )

    if not matching_clusters:

        print("\n===================================================")
        print("NO MATCHING CLUSTERS FOUND")
        print("===================================================")

        print(
            f"\nNo clusters found for support plan: "
            f"{normalized_support_plan}"
        )

        print(
            "\nSkipping org-level billing API call."
        )

        return

    get_aggregated_org_billing(
        session=session,
        org_id=args.org_id,
        matching_clusters=matching_clusters,
        start_date=args.start_date,
        end_date=args.end_date,
        total_only=total_only
    )


if __name__ == "__main__":
    main()