init

fix: add parameter to fetch_metadata() function
KhaiHuy123 · Nov 6, 2024 · 5828a0f · 5828a0f
1 parent 00cd378
commit 5828a0f
Show file tree

Hide file tree

Showing 17 changed files with 18 additions and 18 deletions.
diff --git a/dagster_dbt_processing/assets/constant.py b/dagster_dbt_processing/assets/constant.py
@@ -150,9 +150,9 @@ def review_null_value(df: pl.DataFrame, context: AssetExecutionContext):
     context.log.info(df.null_count().to_arrow())
 
 
-def fetch_metadata(conn):
+def fetch_metadata(conn, table_name):
     metadata = conn.execute(
-        f"SELECT * FROM duckdb_tables() WHERE table_name = '{TABLE_ANL_BASE_REPORT}'"
+        f"SELECT * FROM duckdb_tables() WHERE table_name = '{table_name}'"
     ).fetchdf()
     return metadata
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_base_report.py b/dagster_dbt_processing/assets/ingest_analytics/anl_base_report.py
@@ -104,7 +104,7 @@ def anl_base_report(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=SERVICES_SCHEMA, table=TABLE_ANL_DIM_AGG_BASE_NAME,
                         file_name=TABLE_ANL_DIM_AGG_BASE_NAME.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_BASE_REPORT)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_fhv_vehicle.py b/dagster_dbt_processing/assets/ingest_analytics/anl_fhv_vehicle.py
@@ -151,7 +151,7 @@ def anl_fhv_vehicles(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=SERVICES_SCHEMA, table=TABLE_ANL_DIM_VEHICLE_YEAR_FHV_VEHICLE,
                         file_name=TABLE_ANL_DIM_VEHICLE_YEAR_FHV_VEHICLE.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_FHV_VEHICLE)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_lost_property.py b/dagster_dbt_processing/assets/ingest_analytics/anl_lost_property.py
@@ -119,7 +119,7 @@ def anl_lost_property(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=SERVICES_SCHEMA, table=TABLE_ANL_DIM_LOST_PROPERTY_VEHICLES,
                         file_name=TABLE_ANL_DIM_LOST_PROPERTY_VEHICLES.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_LOST_PROPERTY)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_meter_shop.py b/dagster_dbt_processing/assets/ingest_analytics/anl_meter_shop.py
@@ -87,7 +87,7 @@ def anl_meter_shop(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=REPORT_SCHEMA, table=TABLE_ANL_METER_SHOPS,
                         file_name=TABLE_ANL_METER_SHOPS.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_METER_SHOPS)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_monthly_report.py b/dagster_dbt_processing/assets/ingest_analytics/anl_monthly_report.py
@@ -87,7 +87,7 @@ def anl_monthly_report(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=REPORT_SCHEMA, table=TABLE_ANL_MONTHLY_REPORT,
                         file_name=TABLE_ANL_MONTHLY_REPORT.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_MONTHLY_REPORT)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_new_driver.py b/dagster_dbt_processing/assets/ingest_analytics/anl_new_driver.py
@@ -84,7 +84,7 @@ def anl_new_driver(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=REPORT_SCHEMA, table=TABLE_ANL_NEW_DRIVERS,
                         file_name=TABLE_ANL_NEW_DRIVERS.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_NEW_DRIVERS)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_shl_inspect.py b/dagster_dbt_processing/assets/ingest_analytics/anl_shl_inspect.py
@@ -100,7 +100,7 @@ def anl_shl_inspect(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=SERVICES_SCHEMA, table=TABLE_ANL_DIM_INSPECT_BASE,
                         file_name=TABLE_ANL_DIM_INSPECT_BASE.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_SHL_INSPECT)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_shl_permit.py b/dagster_dbt_processing/assets/ingest_analytics/anl_shl_permit.py
@@ -115,7 +115,7 @@ def anl_shl_permit(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=SERVICES_SCHEMA, table=TABLE_ANL_DIM_BASE,
                         file_name=TABLE_ANL_DIM_BASE.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_SHL_PERMIT)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_analytics/anl_taxi_zone.py b/dagster_dbt_processing/assets/ingest_analytics/anl_taxi_zone.py
@@ -81,7 +81,7 @@ def anl_taxi_zone(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=ZONES_SCHEMA, table=TABLE_ANL_TAXI_ZONES,
                         file_name=TABLE_ANL_TAXI_ZONES.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_ANL_TAXI_ZONES)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_geometry/geo_nta.py b/dagster_dbt_processing/assets/ingest_geometry/geo_nta.py
@@ -75,7 +75,7 @@ def geo_nta(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=ZONES_SCHEMA, table=TABLE_GEO_NTA,
                         file_name=TABLE_GEO_NTA.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_GEO_NTA)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_geometry/geo_taxi_zone.py b/dagster_dbt_processing/assets/ingest_geometry/geo_taxi_zone.py
@@ -75,7 +75,7 @@ def geo_taxi_zone(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=ZONES_SCHEMA, table=TABLE_GEO_TAXI_ZONES,
                         file_name=TABLE_GEO_TAXI_ZONES.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_GEO_TAXI_ZONES)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_time_series/ts_aqe.py b/dagster_dbt_processing/assets/ingest_time_series/ts_aqe.py
@@ -82,7 +82,7 @@ def ts_aqe(context: AssetExecutionContext, duckdb: DuckDBResource):
                         schema=ENV_SCHEMA, table=TABLE_TS_AQE,
                         file_name=TABLE_TS_AQE.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_TS_AQE)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_time_series/ts_fhv_trips.py b/dagster_dbt_processing/assets/ingest_time_series/ts_fhv_trips.py
@@ -83,7 +83,7 @@ def ts_fhv_trips(context: AssetExecutionContext, duckdb: DuckDBResource):
 
             context.log.info(f"Processing {sub_list} finished _ idx : {idx}")
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_TS_FHV_TRIPS)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_time_series/ts_green_trips.py b/dagster_dbt_processing/assets/ingest_time_series/ts_green_trips.py
@@ -118,7 +118,7 @@ def ts_green_trips(context: AssetExecutionContext, duckdb: DuckDBResource):
 
             context.log.info(f"Processing {sub_list} finished _ idx : {idx}")
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_TS_GREEN_TRIPS)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_time_series/ts_hourly_monitoring.py b/dagster_dbt_processing/assets/ingest_time_series/ts_hourly_monitoring.py
@@ -81,7 +81,7 @@ def ts_hourly_monitoring(context: AssetExecutionContext, duckdb: DuckDBResource)
                         schema=ENV_SCHEMA, table=TABLE_TS_HOURLY_MONITORING,
                         file_name=TABLE_TS_HOURLY_MONITORING.lower().replace("_", ""), context=context)
 
-        metadata = fetch_metadata(conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_TS_HOURLY_MONITORING)
 
         metadata_mapping = create_column_value_dict(metadata)
 

diff --git a/dagster_dbt_processing/assets/ingest_time_series/ts_yellow_trips.py b/dagster_dbt_processing/assets/ingest_time_series/ts_yellow_trips.py
@@ -108,7 +108,7 @@ def ts_yellow_trips(context: AssetExecutionContext, duckdb: DuckDBResource):
                             file_name=TABLE_TS_DIM_VENDOR.lower().replace("_", ""), context=context, idx=idx)
             context.log.info(f"Processing {sub_list} finished _ idx : {idx}")
 
-        metadata = fetch_metadata(conn=conn)
+        metadata = fetch_metadata(conn=conn, table_name=TABLE_TS_YELLOW_TRIPS)
 
         metadata_mapping = create_column_value_dict(metadata)