✨ Add Eurostat source and Prefect tasks

src/viadot/orchestration/prefect/flows/__init__.py

@@ -10,6 +10,7 @@
 from .duckdb_to_sql_server import duckdb_to_sql_server
 from .duckdb_transform import duckdb_transform
 from .epicor_to_parquet import epicor_to_parquet
+from .eurostat_to_adls import eurostat_to_adls
 from .exchange_rates_to_adls import exchange_rates_to_adls
 from .exchange_rates_to_databricks import exchange_rates_to_databricks
 from .exchange_rates_to_redshift_spectrum import exchange_rates_api_to_redshift_spectrum
@@ -68,5 +69,6 @@
     "supermetrics_to_adls",
     "transform",
     "transform_and_catalog",
+    "eurostat_to_adls",
     "vid_club_to_adls",
 ]

src/viadot/orchestration/prefect/flows/eurostat_to_adls.py

@@ -0,0 +1,80 @@
+"""Download data from Eurostat and upload it to Azure Data Lake Storage."""
+
+from prefect import flow
+
+from viadot.orchestration.prefect.tasks import df_to_adls, eurostat_to_df
+
+
+@flow(
+    name="extract--eurostat--adls",
+    description="""Flow for downloading data from the Eurostat platform via
+    HTTPS REST API (no credentials required) to a CSV or Parquet file.
+    Then upload it to Azure Data Lake.""",
+    retries=1,
+    retry_delay_seconds=60,
+)
+def eurostat_to_adls(
+    dataset_code: str,
+    adls_path: str,
+    params: dict[str, str] | None = None,
+    columns: list[str] | None = None,
+    tests: dict | None = None,
+    adls_credentials_secret: str | None = None,
+    overwrite_adls: bool = False,
+    adls_config_key: str | None = None,
+) -> None:
+    """Flow for downloading data from Eurostat to Azure Data Lake.
+
+    This module provides a prefect flow function to use the Eurostat connector:
+    - Call the prefect task wrapper to retrieve a DataFrame from the connector.
+    - Upload the retrieved data to Azure Data Lake Storage.
+
+    Args:
+        dataset_code (str):
+            The code of the Eurostat dataset to be uploaded.
+        adls_path (str | None, optional): The destination folder or path
+            in Azure Data Lake.
+        params (dict[str, str] | None, optional):
+            A dictionary with optional URL parameters. Each key is a parameter ID,
+            and the value is a specific parameter code, e.g.,
+            `params = {'unit': 'EUR'}` where "unit" is the parameter, and "EUR"
+            is the code. You can pass one code per parameter. Defaults to None.
+        columns (list[str] | None, optional):
+            A list of columns to filter from the DataFrame after downloading.
+            This acts as a filter to retrieve only the needed columns. Defaults to None.
+        tests (dict | None, optional):
+            A dictionary containing test cases for the data, such as:
+            - `column_size`: dict{column: size}
+            - `column_unique_values`: list[columns]
+            - `column_list_to_match`: list[columns]
+            - `dataset_row_count`: dict{'min': number, 'max': number}
+            - `column_match_regex`: dict{column: 'regex'}
+            - `column_sum`: dict{column: {'min': number, 'max': number}}.
+            Defaults to None.
+
+        adls_credentials_secret (str | None, optional):
+            The Azure Key Vault secret containing Service Principal credentials
+            (TENANT_ID, CLIENT_ID, CLIENT_SECRET) and ACCOUNT_NAME for Azure Data
+            Lake access. Defaults to None.
+        overwrite_adls (bool, optional):
+            Whether to overwrite files in the lake if they exist. Defaults to False.
+        adls_config_key (str | None, optional):
+            The key in the viadot config that holds the credentials for Azure
+            Data Lake. Defaults to None.
+
+    Returns:
+        None
+    """
+    df = eurostat_to_df(
+        dataset_code=dataset_code,
+        params=params,
+        columns=columns,
+        tests=tests,
+    )
+    df_to_adls(
+        df=df,
+        path=adls_path,
+        credentials_secret=adls_credentials_secret,
+        config_key=adls_config_key,
+        overwrite=overwrite_adls,
+    )

src/viadot/orchestration/prefect/tasks/__init__.py

@@ -11,6 +11,7 @@
 from .dbt import dbt_task
 from .duckdb import duckdb_query
 from .epicor import epicor_to_df
+from .eurostat import eurostat_to_df
 from .exchange_rates import exchange_rates_to_df
 from .genesys import genesys_to_df
 from .git import clone_repo
@@ -66,6 +67,7 @@
     "sharepoint_to_df",
     "sql_server_query",
     "sql_server_to_df",
+    "eurostat_to_df",
     "vid_club_to_df",
     "supermetrics_to_df",
 ]

src/viadot/orchestration/prefect/tasks/eurostat.py

@@ -0,0 +1,51 @@
+"""Tasks for interacting with Eurostat."""
+
+import pandas as pd
+from prefect import task
+
+from viadot.sources import Eurostat
+
+
+@task(retries=3, retry_delay_seconds=10, timeout_seconds=60 * 60)
+def eurostat_to_df(
+    dataset_code: str,
+    params: dict[str, str] | None = None,
+    columns: list[str] | None = None,
+    tests: dict | None = None,
+) -> pd.DataFrame:
+    """Task for creating a pandas DataFrame from Eurostat HTTPS REST API.
+
+    This function serves as an intermediate wrapper between the prefect flow
+    and the Eurostat connector:
+    - Instantiates an Eurostat Cloud API connector.
+    - Creates and returns a pandas DataFrame with the response from the API.
+
+    Args:
+        dataset_code (str): The code of the Eurostat dataset that you would like
+            to download.
+        params (dict[str, str] | None, optional):
+            A dictionary with optional URL parameters. Each key is a parameter ID,
+            and the value is a specific parameter code, e.g.,
+            `params = {'unit': 'EUR'}` where "unit" is the parameter, and "EUR" is
+            the code. Only one code per parameter is allowed. Defaults to None.
+        columns (list[str] | None, optional):
+            A list of column names (as strings) that are required from the dataset.
+            Filters the DataFrame to only include the specified columns.
+            Defaults to None.
+        tests (dict | None, optional):
+            A dictionary containing test cases for the data, including:
+            - `column_size`: dict{column: size}
+            - `column_unique_values`: list[columns]
+            - `column_list_to_match`: list[columns]
+            - `dataset_row_count`: dict{'min': number, 'max': number}
+            - `column_match_regex`: dict{column: 'regex'}
+            - `column_sum`: dict{column: {'min': number, 'max': number}}.
+            Defaults to None.
+
+    Returns:
+        pd.DataFrame:
+            A pandas DataFrame containing the data retrieved from the Eurostat API.
+    """
+    return Eurostat(
+        dataset_code=dataset_code, params=params, columns=columns, tests=tests
+    ).to_df()

src/viadot/sources/__init__.py

@@ -10,6 +10,7 @@
 from .cloud_for_customers import CloudForCustomers
 from .customer_gauge import CustomerGauge
 from .epicor import Epicor
+from .eurostat import Eurostat
 from .exchange_rates import ExchangeRates
 from .genesys import Genesys
 from .hubspot import Hubspot
@@ -48,6 +49,7 @@
     "Supermetrics",
     "Trino",
     "UKCarbonIntensity",
+    "Eurostat",
     "VidClub",
 ]