db module, consumer update, etc

Author: spompea-cfa

Dockerfile.consumer

@@ -7,7 +7,7 @@ RUN apt-get update \
  && apt-get -y autoremove \
  && apt-get -y clean
 
-RUN pip3 install --break-system-packages opentelemetry-distro opentelemetry-exporter-otlp-proto-http
+RUN pip3 install --break-system-packages opentelemetry-distro opentelemetry-exporter-otlp-proto-http psycopg2-binary
 # Above, `--break-system-packages` flag overrides the
 # "This environment is externally managed" error that calling pip
 # would otherwise incur here.

dev-scripts/check_export_tracker.py

@@ -1,4 +1,10 @@
+# Usage:
+#   docker compose run tools python dev/check_export_tracker.py
+# To dump contents of export_tracker to stdout, run:
+#   docker compose run tools python dev/check_export_tracker.py dump
+
 import os
+import sys
 import psycopg2
 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
 
@@ -25,6 +31,7 @@
 except Exception as e:
     print(str(e))
 
+rslt = None
 try:
     curs.execute('select * from export_tracker')    
     rslt = curs.fetchall()
@@ -36,3 +43,6 @@
     print(str(e))
 
 print('==================================================')
+
+if len(sys.argv) == 2 and sys.argv[1] == 'dump':
+    print(rslt)

docker-compose.yaml

@@ -96,6 +96,9 @@ services:
     environment:
       AWS_PROFILE: localstack
       AWS_ENDPOINT_URL: http://localstack:4566
+      PGHOST: db
+      PGUSER: ${POSTGRES_USERNAME:-senzing}
+      PGPASSWORD: ${POSTGRES_PASSWORD:-senzing}
       Q_URL: http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/sqs-senzing-local-ingest
       SENZING_ENGINE_CONFIGURATION_JSON: >-
         {
@@ -122,6 +125,9 @@ services:
       db:
         condition: service_healthy
     environment:
+      PGHOST: db
+      PGUSER: ${POSTGRES_USERNAME:-senzing}
+      PGPASSWORD: ${POSTGRES_PASSWORD:-senzing}
       SENZING_ENGINE_CONFIGURATION_JSON: >-
         {
           "PIPELINE": {
@@ -146,6 +152,9 @@ services:
     environment:
       AWS_PROFILE: localstack
       AWS_ENDPOINT_URL: http://localstack:4566
+      PGHOST: db
+      PGUSER: ${POSTGRES_USERNAME:-senzing}
+      PGPASSWORD: ${POSTGRES_PASSWORD:-senzing}
       S3_BUCKET_NAME: sqs-senzing-local-export
       SENZING_ENGINE_CONFIGURATION_JSON: >-
         {

middleware/consumer.py

@@ -13,6 +13,7 @@
 
 import otel
 import util
+import db
 
 try:
     log.info('Importing senzing_core library . . .')
@@ -190,8 +191,12 @@ def clean_up(signum, frm):
                 success_status = otel.SUCCESS
                 log.debug(SZ_TAG + 'Successful add_record having ReceiptHandle: '
                          + receipt_handle)
-                # TODO send affected entity IDs to tracker
+
+                # Save affected entity IDs to tracker table for exporting later.
                 affected = util.parse_affected_entities_resp(resp)
+                log.debug(SZ_TAG + 'Affected entities: ' + str(affected))
+                for entity_id in affected: db.add_entity_id(entity_id)
+
             except KeyError as ke:
                 log.error(fmterr(ke))
                 make_msg_visible(sqs, Q_URL, receipt_handle)

middleware/db.py

@@ -1 +1,46 @@
 import psycopg2
+
+from loglib import *
+log = retrieve_logger()
+
+EXPORT_STATUS_TODO = 1
+EXPORT_STATUS_IN_PROGRESS = 2
+EXPORT_STATUS_DONE = 3
+
+_params = {
+    'dbname': 'sqs_entity_resolution',
+    'user': os.environ['PGUSER'],
+    'password': os.environ['PGPASSWORD'],
+    'host': os.environ['PGHOST'],
+    'port':'5432'}
+
+_conn = psycopg2.connect(**_params)
+_curs = _conn.cursor()
+
+def add_entity_id(entity_id):
+    '''Inserts an entity_id into export_tracker with initial status of
+    EXPORT_STATUS_TODO.'''
+    if type(entity_id) is not int: raise TypeError
+    log.debug(f'Entity ID: {entity_id}')
+    try:
+        _curs.execute(
+            'insert into export_tracker (entity_id, export_status) values (%s, %s)',
+            [entity_id, EXPORT_STATUS_TODO])
+        _conn.commit()
+    except Exception as e:
+        _conn.rollback()
+        log.error(f'Failure to insert for entity ID: {entity_id}')
+
+def tag_todo_as_in_progress_and_retrieve():
+    '''This function does two things:
+    1. For all rows with status of EXPORT_STATUS_TODO, updates them
+       to be EXPORT_STATUS_IN_PROGRESS.
+    2. Returns a *distinct* (no duplicates) list of those entity IDs.'''
+    ...
+
+def tag_in_progress_as_done(export_id):
+    '''For all rows with status of EXPORT_STATUS_IN_PROGRESS, updates them
+    to be EXPORT_STATUS_DONE and collectively updates their export_id value.
+    In Practice, export_id will usually be the name of the output file that
+    was exported.'''
+    ...

sql/create-export-tracker-table.sql

@@ -7,7 +7,7 @@ CREATE DATABASE sqs_entity_resolution
 CREATE TABLE public.export_tracker
 (
     ts timestamp without time zone NOT NULL default current_timestamp,
-    entity_id character varying  NOT NULL,
-    export_status integer NOT NULL DEFAULT 0,
+    entity_id bigint NOT NULL,
+    export_status smallint NOT NULL DEFAULT 0,
     export_id character varying
 )