added in uk charge point data asset and refactored some pydantic models

.gitignore

@@ -186,3 +186,4 @@ test.py
 creds.py
 .streamlit
 poetry.lock
+data_testing/

analytics_platform_dagster/__init__.py

@@ -14,14 +14,16 @@
 )
 from .assets.catalogue_metadata_assets import analytics_platform_datastore_assets
 from .assets.infrastructure_data_assets import (
-    analytics_platform_national_charge_points_assets,
-    uk_power_networks_live_faults
+    national_charge_points_london_assets,
+    uk_power_networks_live_faults,
+    national_charge_points_uk_assets
 )
 
 from .utils.io_manager_helper.io_manager import (
     S3ParquetManager,
     AwsWranglerDeltaLakeIOManager,
     S3JSONManager,
+    PartitionedDuckDBParquetManager
 )
 
 from .jobs.analytics_platfom_jobs import (
@@ -72,9 +74,10 @@ def get_env_var(var_name: str) -> str:
             analytics_platform_ea_flood_public_forecast,
             analytics_platform_dbt_trade_barrier_assets,
             analytics_platform_carbon_intensity_assets,
-            analytics_platform_national_charge_points_assets,
+            national_charge_points_london_assets,
             entsog_uk_gas_assets,
-            uk_power_networks_live_faults
+            uk_power_networks_live_faults,
+            national_charge_points_uk_assets
 
         ]
     ),
@@ -98,6 +101,7 @@ def get_env_var(var_name: str) -> str:
     resources={
         "S3Parquet": S3ParquetManager(bucket_name=get_env_var("BRONZE_DATA_BUCKET")),
         "S3Json": S3JSONManager(bucket_name=get_env_var("BRONZE_DATA_BUCKET")),
+        "PartitionedDuckDBManager": PartitionedDuckDBParquetManager(bucket_name=get_env_var("BRONZE_DATA_BUCKET")),
         "DeltaLake": AwsWranglerDeltaLakeIOManager(
             bucket_name=get_env_var("SILVER_DATA_BUCKET")
         ),

analytics_platform_dagster/assets/infrastructure_data_assets/national_charge_points_london_assets.py

@@ -0,0 +1,78 @@
+"""
+This currently only provides data for London only
+"""
+import pandas as pd
+import io
+
+from pydantic import ValidationError
+from ...utils.requests_helper.requests_helper import return_json
+from ...utils.variables_helper.url_links import asset_urls
+from ...models.infrastructure_data_models.national_charge_point_london_model import (
+    ChargeDevice,
+)
+from dagster import AssetExecutionContext, AssetIn, asset
+from ...utils.slack_messages.slack_message import with_slack_notification
+
+@asset(group_name="infrastructure_assets", io_manager_key="S3Parquet")
+def national_charge_point_london_bronze(context: AssetExecutionContext):
+    """
+    Fetches json data and uploads charge point data to S3 using IO Manager
+
+    Returns:
+        A parquet file in S3
+    """
+
+    # Fetch url
+    url = asset_urls.get("national_charge_point_london")
+    if url is None:
+        raise ValueError("API_ENDPOINT can't be None")
+
+    response = return_json(url)
+    validation_errors = []
+
+
+    try:
+        ChargeDevice.model_validate(response)
+    except ValidationError as e:
+        validation_errors = e.errors()
+
+    df = pd.DataFrame(response)
+    df = df.astype(str)
+
+    context.log.info(f"Processed {len(df)} records with {len(validation_errors)} validation errors")
+
+    parquet_buffer = io.BytesIO()
+    df.to_parquet(parquet_buffer, engine="pyarrow")
+    parquet_bytes = parquet_buffer.getvalue()
+
+    context.log.info("Successfully processed batch into Parquet format")
+    return parquet_bytes
+
+@asset(
+    group_name="infrastructure_assets",
+    io_manager_key="DeltaLake",
+    metadata={"mode": "overwrite"},
+    ins={
+        "national_charge_point_london_bronze": AssetIn(
+            "national_charge_point_london_bronze"
+        )
+    },
+    required_resource_keys={"slack"}
+)
+@with_slack_notification("National EV Charge Point Data London")
+def national_charge_point_london_silver(
+    context: AssetExecutionContext, national_charge_point_london_bronze
+) -> pd.DataFrame:
+    """
+    Write charge point data out to Delta Lake
+
+    Returns:
+        Delta Lake table in S3.
+    """
+    try:
+        df = pd.DataFrame(national_charge_point_london_bronze)
+        return df
+
+    except Exception as e:
+        context.log.error(f"Error processing data: {e}")
+        raise