"""HTTP client for the Cube.js API of Indicateurs Territoriaux.""" import asyncio import os from typing import Any, Literal import httpx from dotenv import load_dotenv load_dotenv() _MAX_ATTEMPTS = 3 _BACKOFF_SECONDS = (0.5, 1.0, 2.0) _RETRYABLE_STATUS = frozenset({502, 503, 504}) class CubeJsClientError(Exception): """Base exception for Cube.js client errors.""" pass class AuthenticationError(CubeJsClientError): """Raised when authentication fails (401).""" pass class BadRequestError(CubeJsClientError): """Raised when the request is malformed (400).""" pass class CubeJsClient: """HTTP client for the Cube.js REST API. This client handles authentication and provides methods to interact with the Indicateurs Territoriaux API endpoints. """ def __init__( self, base_url: str | None = None, token: str | None = None, timeout: float = 30.0, ): """Initialize the Cube.js client. Args: base_url: Base URL of the API. Defaults to env var INDICATEURS_TE_BASE_URL. token: JWT authentication token. Defaults to env var INDICATEURS_TE_TOKEN. timeout: Request timeout in seconds. """ self.base_url = ( base_url or os.getenv("INDICATEURS_TE_BASE_URL") or "https://api.indicateurs.ecologie.gouv.fr" ) self.token = token or os.getenv("INDICATEURS_TE_TOKEN") if not self.token: raise ValueError( "No API token provided. Set INDICATEURS_TE_TOKEN environment variable " "or pass token parameter." ) self.timeout = timeout self._client: httpx.AsyncClient | None = None @property def headers(self) -> dict[str, str]: """HTTP headers for API requests.""" return { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", } async def _get_client(self) -> httpx.AsyncClient: """Get or create the async HTTP client.""" if self._client is None or self._client.is_closed: self._client = httpx.AsyncClient( base_url=self.base_url, headers=self.headers, timeout=self.timeout, ) return self._client async def close(self) -> None: """Close the HTTP client.""" if self._client is not None and not self._client.is_closed: await self._client.aclose() self._client = None async def _handle_response(self, response: httpx.Response) -> dict[str, Any]: """Handle API response and raise appropriate errors. Args: response: The HTTP response object. Returns: Parsed JSON response. Raises: AuthenticationError: If the token is invalid or expired (401). BadRequestError: If the request is malformed (400). CubeJsClientError: For other HTTP errors. """ if response.status_code == 401: raise AuthenticationError( "Authentication failed. Your API token may be invalid or expired. " "Please check your INDICATEURS_TE_TOKEN environment variable." ) if response.status_code == 400: try: error_detail = response.json() except Exception: error_detail = response.text raise BadRequestError( f"Bad request to API. Details: {error_detail}" ) if response.status_code >= 400: raise CubeJsClientError( f"API request failed with status {response.status_code}: {response.text}" ) return response.json() async def _request( self, method: Literal["GET", "POST"], path: str, **kwargs: Any, ) -> dict[str, Any]: client = await self._get_client() for attempt in range(_MAX_ATTEMPTS): try: response = await client.request(method, path, **kwargs) except httpx.TimeoutException as e: if attempt < _MAX_ATTEMPTS - 1: await asyncio.sleep(_BACKOFF_SECONDS[attempt]) continue raise CubeJsClientError( f"Indicators API did not respond within {self.timeout}s" ) from e except httpx.ConnectError as e: if attempt < _MAX_ATTEMPTS - 1: await asyncio.sleep(_BACKOFF_SECONDS[attempt]) continue raise CubeJsClientError("Indicators API unreachable") from e except httpx.HTTPError as e: if attempt < _MAX_ATTEMPTS - 1: await asyncio.sleep(_BACKOFF_SECONDS[attempt]) continue raise CubeJsClientError(f"Indicators API network error: {e}") from e if ( response.status_code in _RETRYABLE_STATUS and attempt < _MAX_ATTEMPTS - 1 ): await asyncio.sleep(_BACKOFF_SECONDS[attempt]) continue return await self._handle_response(response) raise CubeJsClientError("Indicators API unavailable") async def get_meta(self) -> dict[str, Any]: """Fetch the API schema metadata. Returns the complete schema including all cubes, their measures, dimensions, and available filters. Returns: Dict containing the API metadata with 'cubes' key. Raises: AuthenticationError: If authentication fails. CubeJsClientError: For other API errors. """ return await self._request("GET", "/cubejs-api/v1/meta") async def load(self, query: dict[str, Any]) -> dict[str, Any]: """Execute a data query against the Cube.js API. Args: query: The Cube.js query object containing measures, dimensions, filters, and other query parameters. Returns: Dict containing the query results with 'data' key. Raises: AuthenticationError: If authentication fails. BadRequestError: If the query is malformed. CubeJsClientError: For other API errors. Example: >>> query = { ... "measures": ["indicateur_metadata.count"], ... "dimensions": ["indicateur_metadata.id", "indicateur_metadata.libelle"], ... "limit": 10 ... } >>> result = await client.load(query) """ return await self._request( "POST", "/cubejs-api/v1/load", json={"query": query}, ) async def load_indicators_metadata( self, dimensions: list[str] | None = None, filters: list[dict[str, Any]] | None = None, limit: int = 500, ) -> list[dict[str, Any]]: """Load indicator metadata from the indicateur_metadata cube. Convenience method for querying the indicator metadata cube. Args: dimensions: List of dimensions to fetch. Defaults to basic info. filters: Optional list of filters to apply. limit: Maximum number of results. Returns: List of indicator metadata records. """ if dimensions is None: dimensions = [ "indicateur_metadata.id", "indicateur_metadata.libelle", "indicateur_metadata.unite", "indicateur_metadata.description", "indicateur_metadata.mailles_disponibles", "indicateur_metadata.thematique_fnv", "indicateur_metadata.annees_disponibles", ] query: dict[str, Any] = { "dimensions": dimensions, "limit": limit, } if filters: query["filters"] = filters result = await self.load(query) return result.get("data", []) # Dimensions souhaitées pour les sources ; seules celles réellement exposées par # /meta sont demandées (l'API refuse toute dimension inconnue, ex. « libelle »). SOURCE_DIMENSIONS_WANTED = ( "id_indicateur", "nom_source", "libelle", "libelle_source", "description", "producteur_source", "distributeur_source", "license_source", "licence_source", "lien_page", "annees_disponibles_source", "maille_mini_disponible", "couverture_geographique", "date_derniere_extraction", ) async def load_sources_metadata( self, indicator_id: int | None = None, limit: int = 100, available_dimensions: list[str] | None = None, ) -> list[dict[str, Any]]: """Load source metadata from the indicateur_x_source_metadata cube. Args: indicator_id: Optional indicator ID to filter sources. limit: Maximum number of results. available_dimensions: Full dimension names of the cube as listed by /meta. When given, only wanted dimensions that actually exist are requested. Returns: List of source metadata records. """ cube = "indicateur_x_source_metadata" wanted = [f"{cube}.{d}" for d in self.SOURCE_DIMENSIONS_WANTED] if available_dimensions is not None: existing = set(available_dimensions) dimensions = [d for d in wanted if d in existing] else: dimensions = wanted if f"{cube}.id_indicateur" not in dimensions: return [] query: dict[str, Any] = { "dimensions": dimensions, "limit": limit, } if indicator_id is not None: query["filters"] = [ { "member": "indicateur_x_source_metadata.id_indicateur", "operator": "equals", "values": [str(indicator_id)], } ] result = await self.load(query) return result.get("data", []) async def search_indicators_by_libelle( self, search_term: str, limit: int = 50, ) -> list[dict[str, Any]]: """Search indicators by keyword in libelle using contains filter. This uses Cube.js contains operator for server-side filtering. Note: Limited to single term, for multi-term use client-side filtering. Args: search_term: Term to search for in indicator libelle. limit: Maximum number of results. Returns: List of matching indicator metadata records. """ query: dict[str, Any] = { "dimensions": [ "indicateur_metadata.id", "indicateur_metadata.libelle", "indicateur_metadata.description", "indicateur_metadata.unite", "indicateur_metadata.mailles_disponibles", "indicateur_metadata.thematique_fnv", ], "filters": [ { "member": "indicateur_metadata.libelle", "operator": "contains", "values": [search_term], } ], "limit": limit, } result = await self.load(query) return result.get("data", []) async def find_territories( self, cube_name: str, geocode_dim: str, libelle_dim: str, name: str, limit: int = 20, ) -> list[dict[str, Any]]: """Find territories whose name contains `name` (case-insensitive, Cube.js `contains`). Args: cube_name: A data cube that carries the geographic dimensions. geocode_dim: Short dimension name (e.g. "geocode_epci"). libelle_dim: Short dimension name (e.g. "libelle_epci"). name: Text to search in the territory label. limit: Maximum number of distinct territories. Returns: List of rows with the two full dimension names as keys. """ query: dict[str, Any] = { "dimensions": [f"{cube_name}.{geocode_dim}", f"{cube_name}.{libelle_dim}"], "filters": [ { "member": f"{cube_name}.{libelle_dim}", "operator": "contains", "values": [name], } ], "limit": limit, } result = await self.load(query) return result.get("data", []) # Singleton instance for the application _client_instance: CubeJsClient | None = None def get_client() -> CubeJsClient: """Get or create the singleton CubeJsClient instance. Returns: The shared CubeJsClient instance. """ global _client_instance if _client_instance is None: _client_instance = CubeJsClient() return _client_instance async def close_client() -> None: """Close the singleton client instance.""" global _client_instance if _client_instance is not None: await _client_instance.close() _client_instance = None