fix(sdmx): Address all review comments

SandeepTuniki · SandeepTuniki · commit 729437437275 · 2025-09-04T13:01:14.000+05:30
This commit addresses all outstanding review comments by:
- Refactoring the SDMX utilities into an `SdmxClient` class to improve design and remove duplication.
- Sanitizing the `dataflow_id` for use in filenames.
- Modifying the sample scripts to write their output to a dedicated `output/` directory within the samples folder.
- Removing redundant `try...except` blocks from the sample scripts and ensuring non-zero exit codes on failure.
diff --git a/tools/sdmx/dataflow.py b/tools/sdmx/dataflow.py
@@ -1,8 +1,7 @@
 """
 dataflow.py
 
-This module provides reusable, generalized functions to interact with SDMX APIs
-by connecting to a specified REST endpoint.
+This module provides a client class for interacting with SDMX APIs.
 """
 
 import logging
@@ -12,159 +11,101 @@
 from typing import Dict, Any
 
 
-def _get_sdmx_client(agency_id: str, endpoint: str) -> sdmx.Client:
-    """
-    Creates and configures an sdmx.Client for a given agency and endpoint.
-
-    A unique source ID is created for each agency-endpoint combination to avoid
-    caching issues. The source is overridden to ensure the correct endpoint is
-    always used.
-    """
-    source_id = agency_id
-    custom_source = {
-        'id': source_id,
-        'url': endpoint,
-        'name': f'Custom source for {agency_id}'
-    }
-    sdmx.add_source(custom_source, override=True)
-    return sdmx.Client(source_id)
-
-
-def fetch_and_save_metadata(dataflow_id: str, agency_id: str, output_path: str,
-                            endpoint: str):
-    """
-    Fetches the complete metadata for a dataflow and saves the raw
-    SDMX-ML (XML) response to a file.
-
-    Args:
-        dataflow_id (str): The ID of the dataflow (e.g., 'DF_QNA_EXPENDITURE_GROWTH_OECD').
-        agency_id (str): The ID of the agency providing the data (e.g., 'OECD.SDD.NAD').
-        output_path (str): The file path where the raw XML metadata will be saved.
-        endpoint (str): The base URL of the SDMX REST API endpoint.
-
-    Raises:
-        HTTPError: If a network error occurs during the API request.
-        Exception: For other unexpected errors.
-
-    Usage:
-        fetch_and_save_metadata(
-            dataflow_id="DSD_NAMAIN1@DF_QNA_EXPENDITURE_GROWTH_OECD",
-            agency_id="OECD.SDD.NAD",
-            output_path="gdp_growth_metadata.xml",
-            endpoint="https://sdmx.oecd.org/public/rest/"
-        )
-    """
-    try:
-        client = _get_sdmx_client(agency_id, endpoint)
-
-        logging.info(f"Fetching raw metadata for dataflow: {dataflow_id}...")
-
-        flow_msg = client.dataflow(dataflow_id,
-                                   agency_id=agency_id,
-                                   params={'references': 'all'})
-        logging.info(
-            f"Successfully received response from the server: {flow_msg.response.url}"
-        )
-
-        raw_metadata_content = flow_msg.response.text
-        with open(output_path, "w", encoding="utf-8") as f:
-            f.write(raw_metadata_content)
-        logging.info(f"Successfully saved raw metadata to '{output_path}'")
-
-    except HTTPError as e:
-        logging.error(
-            f"Network error while downloading dataflow metadata for {agency_id}/{dataflow_id}: {e}"
-        )
-        if e.response:
-            safe_df_id = dataflow_id.replace('@', '_')
-            error_filename = f"metadata_error_{safe_df_id}.html"
-            with open(error_filename, "w", encoding="utf-8") as f:
-                f.write(e.response.text)
-            logging.error(f"URL: {e.response.url}")
-            logging.error(
-                f"Response content saved to '{error_filename}' for debugging.")
-        raise
-    except Exception as e:
-        logging.error(
-            f"An error occurred while processing dataflow metadata for {agency_id}/{dataflow_id}: {e}"
-        )
-        raise
-
-
-def fetch_and_save_data_as_csv(dataflow_id: str, agency_id: str,
-                               key: Dict[str, Any], params: Dict[str, Any],
-                               output_path: str, endpoint: str):
-    """
-    Fetches data from an SDMX API, converts it to a tidy pandas DataFrame,
-    and saves it as a CSV file.
-
-    Args:
-        dataflow_id (str): The ID of the dataflow.
-        agency_id (str): The ID of the agency.
-        key (dict): A dictionary defining the slice of data to query.
-        params (dict): A dictionary of query parameters (e.g., startPeriod).
-        output_path (str): The file path where the final CSV data will be saved.
-        endpoint (str): The base URL of the SDMX REST API endpoint.
-
-    Raises:
-        HTTPError: If a network error occurs during the API request.
-        Exception: For other unexpected errors.
-
-    Usage:
-        DATA_KEY = {
-            'FREQ': 'Q',
-            'REF_AREA': 'USA+CAN+MEX',
-            'TRANSACTION': 'B1GQ',
-            'TRANSFORMATION': 'G1'
+class SdmxClient:
+    """A client for fetching data and metadata from an SDMX REST API."""
+
+    def __init__(self, endpoint: str, agency_id: str):
+        """
+        Initializes the SdmxClient.
+
+        Args:
+            endpoint (str): The base URL of the SDMX REST API endpoint.
+            agency_id (str): The ID of the agency providing the data.
+        """
+        self.agency_id = agency_id
+        self.endpoint = endpoint
+        self.client = self._get_sdmx_client()
+
+    def _get_sdmx_client(self) -> sdmx.Client:
+        """
+        Creates and configures an sdmx.Client for the specified endpoint and agency.
+        """
+        source_id = self.agency_id
+        custom_source = {
+            'id': source_id,
+            'url': self.endpoint,
+            'name': f'Custom source for {self.agency_id}'
         }
-        DATA_PARAMS = {
-            'startPeriod': '2022',
-            'endPeriod': '2023'
-        }
-        fetch_and_save_data_as_csv(
-            dataflow_id="DSD_NAMAIN1@DF_QNA_EXPENDITURE_GROWTH_OECD",
-            agency_id="OECD.SDD.NAD",
-            key=DATA_KEY,
-            params=DATA_PARAMS,
-            output_path="gdp_growth_data.csv",
-            endpoint="https://sdmx.oecd.org/public/rest/"
-        )
-    """
-    try:
-        client = _get_sdmx_client(agency_id, endpoint)
-
-        logging.info(f"Fetching data for key: {key}")
-
-        data_msg = client.data(dataflow_id,
-                               key=key,
-                               params=params,
-                               agency_id=agency_id)
-        logging.info(
-            f"Successfully received response from the server: {data_msg.response.url}"
-        )
-
-        data_series = sdmx.to_pandas(data_msg)
-        df_tidy = data_series.reset_index()
-
-        df_tidy.to_csv(output_path, index=False)
-        logging.info(
-            f"Successfully converted to CSV data and saved to '{output_path}'")
-
-    except HTTPError as e:
-        logging.error(
-            f"Network error while downloading data for {agency_id}/{dataflow_id}: {e}"
-        )
-        if e.response:
-            safe_df_id = dataflow_id.replace('@', '_')
-            error_filename = f"data_error_{safe_df_id}.html"
-            with open(error_filename, "w", encoding="utf-8") as f:
-                f.write(e.response.text)
-            logging.error(f"URL: {e.response.url}")
+        sdmx.add_source(custom_source, override=True)
+        return sdmx.Client(source_id)
+
+    def fetch_and_save_metadata(self, dataflow_id: str, output_path: str):
+        """
+        Fetches the complete metadata for a dataflow and saves the raw
+        SDMX-ML (XML) response to a file.
+        """
+        try:
+            logging.info(
+                f"Fetching raw metadata for dataflow: {dataflow_id}...")
+            flow_msg = self.client.dataflow(dataflow_id,
+                                            agency_id=self.agency_id,
+                                            params={'references': 'all'})
+            logging.info(
+                f"Successfully received response: {flow_msg.response.url}")
+
+            raw_content = flow_msg.response.text
+            with open(output_path, "w", encoding="utf-8") as f:
+                f.write(raw_content)
+            logging.info(f"Successfully saved metadata to '{output_path}'")
+
+        except HTTPError as e:
+            logging.error(
+                f"Network error for {self.agency_id}/{dataflow_id}: {e}")
+            if e.response:
+                safe_df_id = dataflow_id.replace('@', '_')
+                error_filename = f"metadata_error_{safe_df_id}.html"
+                with open(error_filename, "w", encoding="utf-8") as f:
+                    f.write(e.response.text)
+                logging.error(f"URL: {e.response.url}")
+                logging.error(f"Response saved to '{error_filename}'")
+            raise
+        except Exception as e:
+            logging.error(
+                f"Error processing metadata for {self.agency_id}/{dataflow_id}: {e}"
+            )
+            raise
+
+    def fetch_and_save_data_as_csv(self, dataflow_id: str, key: Dict[str, Any],
+                                   params: Dict[str, Any], output_path: str):
+        """
+        Fetches data, converts it to a pandas DataFrame, and saves as CSV.
+        """
+        try:
+            logging.info(f"Fetching data for key: {key}")
+            data_msg = self.client.data(dataflow_id,
+                                        key=key,
+                                        params=params,
+                                        agency_id=self.agency_id)
+            logging.info(
+                f"Successfully received response: {data_msg.response.url}")
+
+            df = sdmx.to_pandas(data_msg).reset_index()
+            df.to_csv(output_path, index=False)
+            logging.info(f"Successfully saved data to '{output_path}'")
+
+        except HTTPError as e:
+            logging.error(
+                f"Network error for {self.agency_id}/{dataflow_id}: {e}")
+            if e.response:
+                safe_df_id = dataflow_id.replace('@', '_')
+                error_filename = f"data_error_{safe_df_id}.html"
+                with open(error_filename, "w", encoding="utf-8") as f:
+                    f.write(e.response.text)
+                logging.error(f"URL: {e.response.url}")
+                logging.error(f"Response saved to '{error_filename}'")
+            raise
+        except Exception as e:
             logging.error(
-                f"Response content saved to '{error_filename}' for debugging.")
-        raise
-    except Exception as e:
-        logging.error(
-            f"An error occurred while processing data for {agency_id}/{dataflow_id}: {e}"
-        )
-        raise
+                f"Error processing data for {self.agency_id}/{dataflow_id}: {e}"
+            )
+            raise
diff --git a/tools/sdmx/samples/fetch_eurostat_gdp_data.py b/tools/sdmx/samples/fetch_eurostat_gdp_data.py
@@ -13,7 +13,7 @@
 sys.path.append(
     os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
 
-from tools.sdmx import dataflow
+from tools.sdmx.dataflow import SdmxClient
 
 # Configure logging
 logging.basicConfig(level=logging.INFO,
@@ -44,13 +44,12 @@ def main():
 
     logging.info(f"--- Fetching Eurostat Data: {dataflow_id} ---")
 
-    # --- 2. Use the Reusable Function ---
-    dataflow.fetch_and_save_data_as_csv(dataflow_id=dataflow_id,
-                                        agency_id=agency_id,
-                                        key=data_key,
-                                        params=data_params,
-                                        output_path=output_path,
-                                        endpoint=endpoint)
+    # --- 2. Use the SdmxClient ---
+    client = SdmxClient(endpoint, agency_id)
+    client.fetch_and_save_data_as_csv(dataflow_id=dataflow_id,
+                                      key=data_key,
+                                      params=data_params,
+                                      output_path=output_path)
     logging.info(f"--- Successfully downloaded data to {output_path} ---")
 
 
diff --git a/tools/sdmx/samples/fetch_eurostat_gdp_metadata.py b/tools/sdmx/samples/fetch_eurostat_gdp_metadata.py
@@ -13,7 +13,7 @@
 sys.path.append(
     os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
 
-from tools.sdmx import dataflow
+from tools.sdmx.dataflow import SdmxClient
 
 # Configure logging
 logging.basicConfig(level=logging.INFO,
@@ -34,11 +34,10 @@ def main():
 
     logging.info(f"--- Fetching Eurostat Metadata: {dataflow_id} ---")
 
-    # --- 2. Use the Reusable Function ---
-    dataflow.fetch_and_save_metadata(dataflow_id=dataflow_id,
-                                     agency_id=agency_id,
-                                     output_path=output_path,
-                                     endpoint=endpoint)
+    # --- 2. Use the SdmxClient ---
+    client = SdmxClient(endpoint, agency_id)
+    client.fetch_and_save_metadata(dataflow_id=dataflow_id,
+                                   output_path=output_path)
     logging.info(f"--- Successfully downloaded metadata to {output_path} ---")
 
 
diff --git a/tools/sdmx/samples/fetch_oecd_full_gdp_dataset.py b/tools/sdmx/samples/fetch_oecd_full_gdp_dataset.py
@@ -13,7 +13,7 @@
 sys.path.append(
     os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
 
-from tools.sdmx import dataflow
+from tools.sdmx.dataflow import SdmxClient
 
 # Configure logging
 logging.basicConfig(level=logging.INFO,
@@ -34,13 +34,14 @@ def main():
                                         "oecd_gdp_full_metadata.xml")
     data_output_path = os.path.join(output_dir, "oecd_gdp_full_data.csv")
 
-    # --- 2. Fetch Metadata ---
+    # --- 2. Initialize Client ---
+    client = SdmxClient(endpoint, agency_id)
+
+    # --- 3. Fetch Metadata ---
     logging.info("--- Step 1: Starting Metadata Download ---")
     try:
-        dataflow.fetch_and_save_metadata(dataflow_id=dataflow_id,
-                                         agency_id=agency_id,
-                                         output_path=metadata_output_path,
-                                         endpoint=endpoint)
+        client.fetch_and_save_metadata(dataflow_id=dataflow_id,
+                                       output_path=metadata_output_path)
         logging.info(
             f"--- Successfully downloaded metadata to {metadata_output_path} ---"
         )
@@ -49,19 +50,17 @@ def main():
         # Exit with a non-zero status code to indicate failure.
         sys.exit(1)
 
-    # --- 3. Fetch Full Data Series ---
+    # --- 4. Fetch Full Data Series ---
     logging.info("\n--- Step 2: Starting Full Data Download ---")
     # For the full dataset, we use an empty key and no time parameters
     data_key = {}
     data_params = {}
 
     try:
-        dataflow.fetch_and_save_data_as_csv(dataflow_id=dataflow_id,
-                                            agency_id=agency_id,
-                                            key=data_key,
-                                            params=data_params,
-                                            output_path=data_output_path,
-                                            endpoint=endpoint)
+        client.fetch_and_save_data_as_csv(dataflow_id=dataflow_id,
+                                          key=data_key,
+                                          params=data_params,
+                                          output_path=data_output_path)
         logging.info(
             f"--- Successfully downloaded data to {data_output_path} ---")
     except Exception as e:
diff --git a/tools/sdmx/samples/fetch_oecd_gdp_data.py b/tools/sdmx/samples/fetch_oecd_gdp_data.py
diff --git a/tools/sdmx/samples/fetch_oecd_gdp_metadata.py b/tools/sdmx/samples/fetch_oecd_gdp_metadata.py
