Add setup for IAP API connections (#49)

Author: Paul-Joel

eq_cir_proxy_service/main.py

@@ -12,7 +12,7 @@
     exception_404_missing_instrument_id,
     exception_422_invalid_instrument_id,
 )
-from eq_cir_proxy_service.routers import instrument_router
+from eq_cir_proxy_service.routers import instrument
 
 # Load .env file
 load_dotenv(".env")
@@ -57,4 +57,4 @@ async def http_exception_handler(request: Request, exc: StarletteHTTPException)
     )
 
 
-app.include_router(instrument_router.router)
+app.include_router(instrument.router)

eq_cir_proxy_service/routers/instrument_router.py renamed to eq_cir_proxy_service/routers/instrument.py

@@ -7,10 +7,10 @@
 
 from eq_cir_proxy_service.exceptions import exception_messages
 from eq_cir_proxy_service.services.instrument import (
-    instrument_conversion_service,
-    instrument_retrieval_service,
+    conversion,
+    retrieval,
 )
-from eq_cir_proxy_service.services.validators.request_validator import (
+from eq_cir_proxy_service.services.validators.request import (
     validate_version,
 )
 from eq_cir_proxy_service.types.custom_types import Instrument
@@ -35,9 +35,9 @@ async def get_instrument_by_uuid(
         validate_version(version)
         target_version = version
 
-        instrument = await instrument_retrieval_service.retrieve_instrument(instrument_id)
+        instrument = await retrieval.retrieve_instrument(instrument_id)
 
-        return await instrument_conversion_service.convert_instrument(instrument, target_version)
+        return await conversion.convert_instrument(instrument, target_version)
 
     except HTTPException:
         raise  # re-raise so FastAPI handles it properly

eq_cir_proxy_service/services/instrument/instrument_conversion_service.py renamed to eq_cir_proxy_service/services/instrument/conversion.py

@@ -3,12 +3,13 @@
 import os
 
 from fastapi import HTTPException, status
-from httpx import AsyncClient, RequestError
+from httpx import RequestError
 from semver import Version
 from structlog import get_logger
 
 from eq_cir_proxy_service.exceptions import exception_messages
 from eq_cir_proxy_service.types.custom_types import Instrument
+from eq_cir_proxy_service.utils.iap import get_api_client
 
 logger = get_logger()
 
@@ -59,18 +60,8 @@ async def convert_instrument(instrument: Instrument, target_version: str) -> Ins
             target_version=target_version,
         )
 
-        converter_service_base_url = os.getenv("CONVERTER_SERVICE_API_BASE_URL")
         converter_service_endpoint = os.getenv("CONVERTER_SERVICE_CONVERT_CI_ENDPOINT", "/schema")
 
-        if not converter_service_base_url:
-            logger.error("CONVERTER_SERVICE_API_BASE_URL is not configured.")
-            raise HTTPException(
-                status_code=500,
-                detail={
-                    "status": "error",
-                    "message": "CONVERTER_SERVICE_API_BASE_URL configuration is missing.",
-                },
-            )
         if not converter_service_endpoint:
             logger.error("CONVERTER_SERVICE_CONVERT_CI_ENDPOINT is not configured.")
             raise HTTPException(
@@ -81,23 +72,25 @@ async def convert_instrument(instrument: Instrument, target_version: str) -> Ins
                 },
             )
 
-        url = f"{converter_service_base_url}{converter_service_endpoint}"
-        try:
-            async with AsyncClient() as client:
-                response = await client.post(
-                    url,
+        async with get_api_client(
+            url_env="CONVERTER_SERVICE_API_BASE_URL",
+            iap_env="CONVERTER_SERVICE_IAP_CLIENT_ID",
+        ) as converter_service_api_client:
+            try:
+                response = await converter_service_api_client.post(
+                    converter_service_endpoint,
                     json={"instrument": instrument},
                     params={"current_version": current_version, "target_version": target_version},
                 )
-        except RequestError as e:
-            logger.exception("Error occurred while converting instrument: ", error=e)
-            raise HTTPException(
-                status_code=500,
-                detail={
-                    "status": "error",
-                    "message": "Error connecting to Converter Service.",
-                },
-            ) from e
+            except RequestError as e:
+                logger.exception("Error occurred while converting instrument.", error=e)
+                raise HTTPException(
+                    status_code=500,
+                    detail={
+                        "status": "error",
+                        "message": "Error connecting to Converter Service.",
+                    },
+                ) from e
 
         instrument_data: Instrument = response.json()
         return instrument_data

eq_cir_proxy_service/services/instrument/instrument_retrieval_service.py renamed to eq_cir_proxy_service/services/instrument/retrieval.py

@@ -4,14 +4,15 @@
 from uuid import UUID
 
 from fastapi import HTTPException
-from httpx import AsyncClient, RequestError
+from httpx import RequestError
 from structlog import get_logger
 
 from eq_cir_proxy_service.exceptions.exception_messages import (
     EXCEPTION_404_INSTRUMENT_NOT_FOUND,
     EXCEPTION_500_INSTRUMENT_PROCESSING,
 )
 from eq_cir_proxy_service.types.custom_types import Instrument
+from eq_cir_proxy_service.utils.iap import get_api_client
 
 logger = get_logger()
 
@@ -27,18 +28,8 @@ async def retrieve_instrument(instrument_id: UUID) -> Instrument:
     """
     logger.debug("Retrieving instrument from CIR...", instrument_id=instrument_id)
 
-    cir_base_url = os.getenv("CIR_API_BASE_URL")
     cir_endpoint = os.getenv("CIR_RETRIEVE_CI_ENDPOINT", "/v2/retrieve_collection_instrument")
 
-    if not cir_base_url:
-        logger.error("CIR_API_BASE_URL is not configured.")
-        raise HTTPException(
-            status_code=500,
-            detail={
-                "status": "error",
-                "message": "CIR_API_BASE_URL configuration is missing.",
-            },
-        )
     if not cir_endpoint:
         logger.error("CIR_RETRIEVE_CI_ENDPOINT is not configured.")
         raise HTTPException(
@@ -49,19 +40,21 @@ async def retrieve_instrument(instrument_id: UUID) -> Instrument:
             },
         )
 
-    url = f"{cir_base_url}{cir_endpoint}"
-    try:
-        async with AsyncClient() as client:
-            response = await client.get(url, params={"guid": str(instrument_id)})
-    except RequestError as e:
-        logger.exception("Error occurred while retrieving instrument.", error=e)
-        raise HTTPException(
-            status_code=500,
-            detail={
-                "status": "error",
-                "message": "Error connecting to CIR service.",
-            },
-        ) from e
+    async with get_api_client(
+        url_env="CIR_API_BASE_URL",
+        iap_env="CIR_IAP_CLIENT_ID",
+    ) as cir_api_client:
+        try:
+            response = await cir_api_client.get(cir_endpoint, params={"guid": str(instrument_id)})
+        except RequestError as e:
+            logger.exception("Error occurred while retrieving instrument.", error=e)
+            raise HTTPException(
+                status_code=500,
+                detail={
+                    "status": "error",
+                    "message": "Error connecting to CIR service.",
+                },
+            ) from e
 
     if response.status_code == 200:
         logger.info("Instrument retrieved successfully.", instrument_id=instrument_id)

eq_cir_proxy_service/services/validators/request_validator.py renamed to eq_cir_proxy_service/services/validators/request.py

@@ -0,0 +1,58 @@
+"""Utility functions for handling IAP authentication and HTTP clients."""
+
+import os
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import google.oauth2.id_token
+from google.auth.transport import requests
+from httpx import AsyncClient
+from structlog import get_logger
+
+logger = get_logger()
+
+
+def get_iap_token(audience: str) -> str:
+    """Fetch an ID token for the IAP-secured resource (blocking)."""
+    token: str = google.oauth2.id_token.fetch_id_token(requests.Request(), audience)  # type: ignore[no-untyped-call]
+    if token is None:
+        logger.error("Failed to fetch IAP token", audience=audience)
+        error_message = f"Failed to fetch IAP token for audience {audience}"
+        raise RuntimeError(error_message)
+    return token
+
+
+@asynccontextmanager
+async def get_api_client(*, url_env: str, iap_env: str) -> AsyncIterator[AsyncClient]:
+    """Context-managed httpx.AsyncClient that switches between IAP and non-IAP connections.
+
+    Args:
+        url_env (str): Environment variable holding the base URL of the API.
+        iap_env (str): Environment variable holding the IAP client ID of the API.
+
+    Raises:
+        RuntimeError: If the base URL environment variable is missing or empty.
+
+    Yields:
+        httpx.AsyncClient: An httpx.AsyncClient instance.
+    """
+    base_url = os.getenv(url_env)
+    audience = os.getenv(iap_env)
+
+    if not base_url:
+        logger.error("Missing or empty environment variable for GCP base URL", var=url_env)
+        base_url_error = f"Missing or empty environment variable: {url_env}"
+        raise RuntimeError(base_url_error)
+
+    if audience:
+        logger.info("Using GCP API client", url_env=url_env, iap_env=iap_env)
+        client = AsyncClient(
+            base_url=base_url,
+            headers={"Authorization": f"Bearer {get_iap_token(audience)}"},
+        )
+    else:
+        logger.info("No IAP client ID set. Using local API client.")
+        client = AsyncClient(base_url=base_url)
+
+    yield client
+    await client.aclose()