Merge branch 'main' into di-543-govnotify-ingestion-for-customer-services

Author: annajgibson

lambdas/lambda_layers/requirements4.txt

@@ -1 +1 @@
-numpy==2.2.3
+numpy==2.2.4

lambdas/lambda_layers/requirements5.txt

@@ -1,4 +1,4 @@
 gspread==6.2.0
 oauth2client==4.1.3
-google-api-python-client==2.163.0
+google-api-python-client==2.166.0
 yagmail==0.15.293

lambdas/lambda_layers/requirements7.txt

@@ -1 +1 @@
-s3fs==2025.3.0
+s3fs==2025.3.2

scripts/jobs/parking/parking_permit_street_cpz_stress_mc.py

@@ -24,6 +24,7 @@
  11/02/2025 - add totals for following permit types 
      permit_type, All Zone, Doctor, Dispensation, Health and Social Care
      Leisure centre permit, Community Support Permit
+01/04/2025 - add check for is_motorcycle is NULL
  *********************************************************************************/
 With parking_permit_denormalised_data as (
     SELECT *
@@ -155,8 +156,14 @@
         street_name
 ),
 /*** get the VRM details to obtain the m/c flag ***/
+/*** 01-04-2025 - tidy the code ***/ 
 MC_Permit_Flag as (
-    SELECT *
+     SELECT permit_reference, vrm, make, model, fuel, engine_capactiy, co2_emission, colour, 
+            foreign, door_plan, lpg_conversion, 
+            CASE 
+                When length(ltrim(rtrim(is_motorcycle))) = 0 Then 'N'
+                ELSE ltrim(rtrim(is_motorcycle))
+            END as is_motorcycle
     from "dataplatform-prod-liberator-raw-zone".liberator_permit_vrm_480
     WHERE import_date = format_datetime(current_date, 'yyyyMMdd')
 ),
@@ -193,14 +200,14 @@
         AND Permit_type not IN ('Suspension')
         AND latest_permit_status not IN ('Cancelled', 'Rejected', 'RENEW_REJECTED')
 ),
+ /*** 01-04-2025 - add check for is_motorcycle is NULL ***/
 Permits_summ as (
     SELECT permit_type,
         cpz,
         address2 as Street,
         cast(count(*) as int) as CurrentPermitTotal
     FROM Permits
-    WHERE r1 = 1
-        AND is_motorcycle != 'Y'
+    WHERE r1 = 1 AND is_motorcycle is NULL OR rtrim(ltrim(is_motorcycle)) = 'N'
     GROUP BY permit_type,
         cpz,
         address2

scripts/jobs/planning/tascomi_create_daily_snapshot.py

@@ -82,6 +82,29 @@ def loadIncrementsSinceDate(increment_table_name, name_space, date):
     return increment_df
 
 
+def purge_today_partition(
+    glueContext: GlueContext, target_destination: str, retentionPeriod: int = 0
+) -> None:
+    """
+    Purges (delete) only today's partition under the given target destination.
+    Parameters:
+      glueContext: GlueContext instance.
+      target_destination: Base S3 path (e.g., "s3://your-bucket/path").
+      retentionPeriod: Retention period in hours (default 0, meaning delete all files immediately).
+    Returns:
+      partition_path: The S3 partition path that was purged.
+    """
+    now = datetime.now()
+    snapshot_year = str(now.year)
+    snapshot_month = str(now.month).zfill(2)
+    snapshot_day = str(now.day).zfill(2)
+    snapshot_date = snapshot_year + snapshot_month + snapshot_day
+
+    partition_path = f"{target_destination}/snapshot_year={snapshot_year}/snapshot_month={snapshot_month}/snapshot_day={snapshot_day}/snapshot_date={snapshot_date}"
+
+    glueContext.purge_s3_path(partition_path, {"retentionPeriod": retentionPeriod})
+
+
 # dict containing parameters for DQ checks
 dq_params = {
     "appeals": {"unique": ["id"]},
@@ -189,7 +212,7 @@ def loadIncrementsSinceDate(increment_table_name, name_space, date):
             # snapshot table in glue catalogue
             else:
                 pushDownPredicate = create_pushdown_predicate(
-                    partitionDateColumn="snapshot_date", daysBuffer=3
+                    partitionDateColumn="snapshot_date", daysBuffer=30
                 )
                 #   load latest snpashot
                 snapshot_ddf = glueContext.create_dynamic_frame.from_catalog(
@@ -318,6 +341,9 @@ def loadIncrementsSinceDate(increment_table_name, name_space, date):
                     snapshot_df, glueContext, "resultDataFrame"
                 )
                 target_destination = s3_bucket_target + table_name
+
+                # Clean up today's partition before writing
+                purge_today_partition(glueContext, target_destination)
                 parquetData = glueContext.write_dynamic_frame.from_options(
                     frame=resultDataFrame,
                     connection_type="s3",
@@ -331,5 +357,6 @@ def loadIncrementsSinceDate(increment_table_name, name_space, date):
     finally:
         if len(dq_errors) > 0:
             logger.error(f"DQ Errors: {dq_errors}")
+            raise Exception(f"Data quality check failed: {'; '.join(dq_errors)}")
         spark.sparkContext._gateway.close()
         spark.stop()