Refactoring ETL folder, preparing first draft of etl_post and etl_load functions.

Author: Milo Hyben

etl/README.md

@@ -0,0 +1,125 @@
+# ETL
+
+Cloud Functions for ETL.
+
+## ETL_POST
+
+etl_post function accepts json payload. It will insert new record in BigQuery table defined by env variable BIGQUERY_TABLE and pushes new message to PUBSUB_TOPIC
+
+
+## ETL_LOAD
+
+etl_load function expects "request_id" in the payload. It is setup as push subscriber to PUBSUB_TOPIC.
+
+
+## How to test locally
+
+### 1. Create BQ table "<YOUR-PROJECT-NAME>.metamist.etl-data"
+  
+### 2. Create TOPIC "projects/<YOUR-PROJECT-NAME>/topics/etl-topic"
+
+### 3. Setup your environment
+
+```bash
+# setup authentication
+gcloud auth application-default login
+export GOOGLE_APPLICATION_CREDENTIALS='/Users/<USERNAME>/.config/gcloud/application_default_credentials.json'
+
+export BIGQUERY_TABLE=<YOUR-PROJECT-NAME>.metamist.etl-data
+export PUBSUB_TOPIC='projects/<YOUR-PROJECT-NAME>/topics/etl-topic'
+```
+
+### 4. Setup python env
+
+```bash
+cd post
+python3 -m venv env
+source env/bin/activate
+pip install -r requirements.txt
+```
+
+### 5. Start POST Fun locally
+
+```bash
+functions-framework-python --target etl_post --debug
+```
+
+### 6. Call etl_post
+
+```bash
+curl -X 'POST' \
+  'http://localhost:8080/' \
+  -H 'Content-Type: application/json' \
+  -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
+  -d '{"identifier": "AB0002", "name": "j smith", "age": 50, "measurement": "98.7", "observation": "B++", "receipt_date": "1/02/2023"}'
+```
+
+Should return something like this:
+
+```bash
+{
+  "id": "76263e55-a869-4604-afe2-441d9c20221e",
+  "success": true
+}
+```
+
+### 7. Start LOAD Fun locally
+Repeat Step 4 inside folder load
+
+```bash
+functions-framework-python --target etl_load --debug
+```
+
+### 7. Call etl_load
+
+Replace request_id with the id returned in Step 6
+
+```bash
+curl -X 'POST' \
+  'http://localhost:8080/' \
+  -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
+  -d '{"request_id": "76263e55-a869-4604-afe2-441d9c20221e"}'
+```
+
+Should return something like this:
+
+```bash
+{
+  "id": "76263e55-a869-4604-afe2-441d9c20221e",
+  "success": true
+}
+```
+
+
+### 8. Deploy functions for testing on the cloud
+
+```bash
+cd ../load
+
+gcloud functions deploy etl_load \
+    --gen2 \
+    --runtime=python311 \
+    --project=<YOUR-PROJECT-NAME> \
+    --region=australia-southeast1 \
+    --source=. \
+    --entry-point=etl_load \
+    --trigger-http \
+    --set-env-vars BIGQUERY_TABLE='<YOUR-PROJECT-NAME>.metamist.etl-data'
+```
+
+```bash
+cd ../post
+
+gcloud functions deploy etl_post \
+    --gen2 \
+    --runtime=python311 \
+    --project=<YOUR-PROJECT-NAME> \
+    --region=australia-southeast1 \
+    --source=. \
+    --entry-point=etl_post \
+    --trigger-http \
+    --set-env-vars BIGQUERY_TABLE=<YOUR-PROJECT-NAME>.metamist.etl-data' \
+    --set-env-vars PUBSUB_TOPIC='projects/<YOUR-PROJECT-NAME>/topics/my-topic'
+```
+
+

etl/endpoint/__init__.py renamed to etl/load/__init__.py

@@ -0,0 +1,79 @@
+import datetime
+import json
+import logging
+import os
+import uuid
+import functions_framework
+import flask
+import google.cloud.bigquery as bq
+
+BIGQUERY_TABLE = os.getenv('BIGQUERY_TABLE')
+
+_BQ_CLIENT = bq.Client()
+
+
+@functions_framework.http
+def etl_load(request: flask.Request):
+    """HTTP Cloud Function.
+    Args:
+        request (flask.Request): The request object.
+        <https://flask.palletsprojects.com/en/1.1.x/api/#incoming-request-data>
+    Returns:
+        The response text, or any set of values that can be turned into a
+        Response object using `make_response`
+        <https://flask.palletsprojects.com/en/1.1.x/api/#flask.make_response>.
+    Note:
+        For more information on how Flask integrates with Cloud
+        Functions, see the `Writing HTTP functions` page.
+        <https://cloud.google.com/functions/docs/writing/http#http_frameworks>
+        
+    This function accepts Pub/Sub message, expected request_id in the payload:
+    
+    {
+        "request_id": "70eb6292-6311-44cf-9c9b-2b38bb076699"
+    }
+    
+    """
+    
+    auth = request.authorization
+    if not auth or not auth.token:
+        return {'success': False, 'message': 'No auth token provided'}, 401
+    
+    # if mimetype might not be set esp. when PubSub pushing from another topic, 
+    # try to force conversion and if fails just return None
+    jbody = request.get_json(force=True, silent=True)
+    
+    if callable(jbody):
+        # request.json is it in reality, but the type checker is saying it's callable
+        jbody = jbody()
+
+    request_id = jbody.get('request_id', None)
+    # check if request_id present
+    if not request_id:
+        jbody_str = json.dumps(jbody)
+        return {
+            'success': False,
+            'message': f'Missing or empty request_id: {jbody_str}',
+        }, 400
+    
+    # locate the request_id in bq
+    query = f"""
+		SELECT * FROM `{BIGQUERY_TABLE}` WHERE request_id = @request_id
+    """
+    query_params = [
+        bq.ScalarQueryParameter("request_id", "STRING", request_id),
+    ]
+    
+    job_config = bq.QueryJobConfig()
+    job_config.query_parameters = query_params
+    query_job = _BQ_CLIENT.query(query, job_config=job_config).result()
+    
+    # should be only one record, look into loading multiple objects in one call?
+    for row in query_job:
+        # TODO
+        # Parse row.body -> Model and upload to metamist database
+        row_body = json.loads(row.body)
+        logging.info(f'row_body {row_body}')
+        
+        
+    return {'id': request_id, 'success': True}

etl/load/main.py

@@ -0,0 +1,4 @@
+flask
+functions_framework
+google-cloud-bigquery
+google-cloud-logging

etl/load/requirements.txt

@@ -10,6 +10,7 @@
 
 from cpg_utils.cloud import email_from_id_token
 
+
 BIGQUERY_TABLE = os.getenv('BIGQUERY_TABLE')
 PUBSUB_TOPIC = os.getenv('PUBSUB_TOPIC')
 
@@ -72,8 +73,10 @@ def etl_post(request: flask.Request):
         }, 500
 
     # publish to pubsub
+    # message contains all the attributes except body which can be large and already stored in BQ table
+    pb_obj = {x: bq_obj[x] for x in bq_obj if x not in ['body']}
     try:
-        _PUBSUB_CLIENT.publish(PUBSUB_TOPIC, json.dumps(bq_obj).encode())
+        _PUBSUB_CLIENT.publish(PUBSUB_TOPIC, json.dumps(pb_obj).encode(), content_type='application/json')
     except Exception as e:  # pylint: disable=broad-exception-caught
         logging.error(f'Failed to publish to pubsub: {e}')