Apispecloader - loads an OpenAPI Spec from a file or from API Hub #py

apispecloader.py

# Copyright © 2025 Google LLC.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import base64
import os
import re

import requests


def is_running_in_cloud_run():
    """Returns True if running in Cloud Run."""
    return bool(os.environ.get("K_SERVICE"))


def _get_gcp_token():
    """Retrieves an OAuth2 token from the GCP metadata server."""
    url = "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token"
    headers = {"Metadata-Flavor": "Google"}
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()["access_token"]


def get_openapi_spec():
    """Retrieves the OpenAPI spec from API Hub, or from a local file."""

    if is_running_in_cloud_run():
        hub = "https://apihub.googleapis.com"
        proj = os.environ.get("APIHUB_PROJ")
        loc = os.environ.get("APIHUB_LOC")
        api = os.environ.get("APIHUB_API")
        version = os.environ.get("APIHUB_API_VER")
        spec = os.environ.get("APIHUB_API_VER_SPEC")
        endpoint = os.environ.get("API_ENDPOINT")

        api_hub_vars = {
            "APIHUB_PROJ": proj,
            "APIHUB_LOC": loc,
            "APIHUB_API": api,
            "APIHUB_API_VER": version,
            "APIHUB_API_VER_SPEC": spec,
            "API_ENDPOINT": endpoint,
        }
        missing_vars = [key for key, value in api_hub_vars.items() if not value]
        if missing_vars:
            raise ValueError(
                "The following required API Hub environment variables are not set: "
                f"{', '.join(missing_vars)}"
            )

        url = (
            f"{hub}/v1/projects/{proj}/locations/{loc}/apis/{api}/versions/{version}/specs/"
            f"{spec}:contents"
        )

        token = _get_gcp_token()
        headers = {"Authorization": f"Bearer {token}"}
        response = requests.get(url, headers=headers, timeout=30)
        response.raise_for_status()
        spec_string = base64.b64decode(response.json()["contents"]).decode("utf-8")
        spec_string = re.sub(
            r"^\s*-\s*url:.*$",
            f"  - url: {endpoint}",
            spec_string,
            flags=re.MULTILINE,
        )
        return spec_string
    else:
        with open(
            os.path.join(os.path.dirname(__file__), "apispec1.yaml"),
            "r",
            encoding="utf-8",
        ) as f:
            return f.read()