#36 optionally send copy to jdbc

Dockerfile

@@ -13,7 +13,7 @@
 # test via (provide payload):
 # curl http://localhost:9000/2015-03-31/functions/function/invocations -d "{payload}"
 #
-# Deploy to AWS Lambda via ACR
+# Deploy to AWS Lambda via ECR
 
 FROM --platform=linux/arm64 public.ecr.aws/lambda/python:3.13-arm64
 
@@ -53,7 +53,7 @@ RUN \
   echo "###################" && \
   echo "### pip installs ###" && \
   echo "###################" && \
-    pip install requests==2.31.0 urllib3==1.26.18 setuptools cryptography jsonschema PyJWT && \
+    pip install requests==2.31.0 urllib3==1.26.18 setuptools cryptography jsonschema PyJWT psycopg2-binary && \
   echo "######################" && \
   echo "### confluent-kafka ###" && \
   echo "######################" && \
@@ -71,7 +71,7 @@ RUN \
 
 # Lambda and SASL_SSL_Artifacts
 COPY $SASL_SSL_ARTIFACTS /opt/sasl_ssl_artifacts/
-COPY src/event_gate_lambda.py $LAMBDA_TASK_ROOT
+COPY src/ $LAMBDA_TASK_ROOT/
 COPY conf $LAMBDA_TASK_ROOT/conf
 
 # Mark librdkafka to LD_LIBRARY_PATH  

conf/access.json

@@ -1,8 +1,8 @@
 {
-    "run.topic": [
+    "public.cps.za.runs": [
         "FooBarUser"
     ],
-    "edla.change.topic": [
+    "public.cps.za.dlchange": [
         "FooUser",
         "BarUser"
     ]

conf/config.json

@@ -1,6 +1,5 @@
 {
     "access_config": "s3://<redacted>/access.json",
-    "topics_config": "s3://<redacted>/topics.json",
     "token_provider_url": "https://<redacted>",
     "token_public_key_url": "https://<redacted>",
     "kafka_bootstrap_server": "localhost:9092",

conf/topic_dlchange.json

@@ -0,0 +1,55 @@
+{
+    "type": "object",
+    "properties": {
+        "event_id": {
+            "type": "string",
+            "description": "Unique identifier for the event (GUID)"
+        },
+        "tenant_id": {
+            "type": "string",
+            "description": "Application ID or ServiceNow identifier"
+        },
+        "source_app": {
+            "type": "string",
+            "description": " Standardized source application name (aqueduct, unify, lum, etc)"
+        },
+        "source_app_version": {
+            "type": "string",
+            "description": "Source application version (SemVer preferred)"
+        },
+        "environment": {
+            "type": "string",
+            "description": "Environment (dev, uat, pre-prod, prod, test or others)"
+        },
+        "timestamp_event": {
+            "type": "number",
+            "description": "Timestamp of the event in epoch milliseconds"
+        },
+        "catalog_id": {
+            "type": "string",
+            "description": "Identifier for the data definition (Glue/Hive) database and table name for example "
+        },
+        "operation": {
+            "type": "string",
+            "enum": ["overwrite", "append", "archive", "delete"],
+            "description": "Operation performed"
+        },
+        "location": {
+            "type": "string",
+            "description": "Location of the data"
+        },
+        "format": {
+            "type": "string",
+            "description": "Format of the data (parquet, delta, crunch, etc)."
+        },
+        "format_options": {
+            "type": "object",
+            "description": "When possible, add additional options related to the format"
+        },
+        "additional_info": {
+            "type": "object",
+            "description": "Optional additional fields structured as an inner JSON"
+        }
+    },
+    "required": ["event_id", "tenant_id", "source_app", "source_app_version", "environment", "timestamp_event", "catalog_id", "operation", "format"]
+}

conf/topic_runs.json

@@ -0,0 +1,73 @@
+{
+    "type": "object",
+    "properties": {
+        "event_id": {
+            "type": "string",
+            "description": "Unique identifier for the event (GUID), generated for each unique event, for de-duplication purposes"
+        },
+        "job_ref": {
+            "type": "string",
+            "description": "Identifier of the job in it’s respective system (e.g. Spark Application Id, Glue Job Id, EMR Step Id, etc)."
+        },
+        "tenant_id ": {
+            "type": "string",
+            "description": "Application ID (4 letter code) or ServiceNow identifier related to the pipeline/domain/process owner (tenant of the tool)"
+        },
+        "source_app": {
+            "type": "string",
+            "description": "Standardized source application name (aqueduct, unify, lum, etc)"
+        },
+        "source_app_version": {
+            "type": "string",
+            "description": "Source application version (SemVer preferred)"
+        },
+        "environment": {
+            "type": "string",
+            "description": "Environment (dev, uat, pre-prod, prod, test or others)"
+        },
+        "timestamp_start": {
+            "type": "number",
+            "description": "Start timestamp of the run in epoch milliseconds"
+        },
+        "timestamp_end": {
+            "type": "number",
+            "description": "End timestamp of the run in epoch milliseconds"
+        },
+        "jobs": {
+            "type": "array",
+            "description": "List of individual jobs withing the run",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "catalog_id": {
+                        "type": "string",
+                        "description": "Identifier for the data definition (Glue/Hive) database and table name for  example"
+                    },
+                    "status": {
+                        "type": "string",
+                        "enum": ["succeeded", "failed", "killed", "skipped"],
+                        "description": "Status of the job."
+                    },
+                    "timestamp_start": {
+                        "type": "number",
+                        "description": "Start timestamp of a job that is a part of a run in epoch milliseconds"
+                    },
+                    "timestamp_end": {
+                        "type": "number",
+                        "description": "End timestamp of a job that is a part of a run in epoch milliseconds"
+                    },
+                    "message": {
+                        "type": "string",
+                        "description": "Job status/error message."
+                    },
+                    "additional_info": {
+                        "type": "object",
+                        "description": "Optional additional fields structured as an inner JSON"
+                    }
+                },
+                "required": ["catalog_id", "status", "timestamp_start", "timestamp_end"]
+            }
+        }
+    },
+    "required": ["event_id", "job_ref", "tenant_id", "source_app", "source_app_version", "environment", "timestamp_start", "timestamp_end", "jobs"]
+}