feat: Exporter middleware

Author: spompea-cfa

Dockerfile.exporter

@@ -0,0 +1,22 @@
+FROM senzing/senzingsdk-runtime:4.0.0
+
+USER root
+
+RUN apt-get update \
+ && apt-get -y install --no-install-recommends curl python3 python3-pip python3-boto3 \
+ && apt-get -y autoremove \
+ && apt-get -y clean
+
+WORKDIR /app
+COPY middleware/* .
+
+# Add a new user and switch to it.
+RUN useradd -m -u 1001 senzing
+USER senzing
+
+ENV PYTHONPATH=/opt/senzing/er/sdk/python:/app
+
+# Flush buffer - helps with print statements.
+ENV PYTHONUNBUFFERED=1
+
+CMD ["python3", "exporter.py"]

docker-compose.yaml

@@ -100,6 +100,31 @@ services:
       # Note: `.aws` mount might not be needed later.
       - ~/.aws:/home/senzing/.aws
 
+  exporter:
+    build:
+      context: .
+      dockerfile: Dockerfile.exporter
+    depends_on:
+      - db
+    environment:
+      AWS_ENDPOINT_URL: http://localstack:4566
+      SENZING_ENGINE_CONFIGURATION_JSON: >-
+        {
+          "PIPELINE": {
+            "CONFIGPATH": "/etc/opt/senzing",
+            "LICENSESTRINGBASE64": "${SENZING_LICENSE_BASE64_ENCODED}",
+            "RESOURCEPATH": "/opt/senzing/er/resources",
+            "SUPPORTPATH": "/opt/senzing/data"
+          },
+          "SQL": {
+            "BACKEND": "SQL",
+            "CONNECTION": "postgresql://${POSTGRES_USERNAME:-senzing}:${POSTGRES_PASSWORD:-senzing}@db:5432:${POSTGRES_DB:-G2}/?sslmode=disable"
+          }
+        }
+    volumes:
+      - ~/.aws:/home/senzing/.aws
+      - ~/tmp:/tmp # Should you wish to write files to host.
+
 volumes:
   localstack:
   postgres:

middleware/exporter.py

@@ -0,0 +1,124 @@
+import json
+import io
+import os
+import time
+import sys
+import boto3
+import senzing as sz
+
+from loglib import *
+log = retrieve_logger()
+
+try:
+    log.info('Importing senzing_core library . . .')
+    import senzing_core as sz_core
+    log.info('Imported senzing_core successfully.')
+except Exception as e:
+    log.error('Importing senzing_core library failed.')
+    log.error(e)
+    sys.exit(1)
+
+SZ_CONFIG = json.loads(os.environ['SENZING_ENGINE_CONFIGURATION_JSON'])
+S3_BUCKET = 'sqs-senzing-local-export'
+
+# TODO which flags do we need?
+EXPORT_FLAGS =  sz.SzEngineFlags.SZ_EXPORT_DEFAULT_FLAGS
+
+#-------------------------------------------------------------------------------
+
+def ts():
+    '''Return current timestamp in ms as a str'''
+    return str(int(round(time.time() * 1000)))
+
+def make_s3_client():
+    sess = boto3.Session()
+    if 'AWS_ENDPOINT_URL'in os.environ:
+        return sess.client('s3', endpoint_url=os.environ['AWS_ENDPOINT_URL'])
+    else:
+        return sess.client('s3')
+
+def go():
+    '''
+    Exports Senzing JSON entity report data into a buffer, then
+    uploads the buffer as a file into the output S3 bucket.
+
+    References:
+    - https://garage.senzing.com/sz-sdk-python/senzing.html#senzing.szengine.SzEngine.export_json_entity_report
+    - https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/upload_fileobj.html
+    '''
+
+    # Init S3 client
+    s3 = make_s3_client()
+    
+    # Init senzing engine object.
+    # Note that Senzing engine object cannot be passed around between functions,
+    # else it will be eagerly cleaned up / destroyed and no longer usable. 
+    sz_eng = None
+    try:
+        sz_factory = sz_core.SzAbstractFactoryCore("ERS", SZ_CONFIG)
+        sz_eng = sz_factory.create_engine()
+        log.info(SZ_TAG + 'Senzing engine object instantiated.')
+    except sz.SzError as sz_err:
+        log.error(SZ_TAG + str(sz_err))
+        sys.exit(1)
+    except Exception as e:
+        log.error(str(e))
+        sys.exit(1)
+
+    # init buffer
+    buff = io.BytesIO()
+    
+    # retrieve output from sz into buff
+    try:
+        export_handle = sz_eng.export_json_entity_report(EXPORT_FLAGS)
+        while 1:
+            chunk = sz_eng.fetch_next(export_handle)
+            if not chunk:
+                break
+            buff.write(chunk.encode('utf-8'))
+        sz_eng.close_export_report(export_handle)
+    except sz.SzError as err:
+        print(err)
+
+    # rewind buffer
+    buff.seek(0)
+
+    # write buff to S3 using upload_fileobj
+    fname = 'output-' + ts() + '.json'
+    resp = s3.upload_fileobj(buff, S3_BUCKET, fname)
+
+    print(resp)
+    return resp
+
+#-------------------------------------------------------------------------------
+
+def main():
+    log.info('====================')
+    log.info('     EXPORTER')
+    log.info('     *STARTED*')
+    log.info('====================')
+    go()
+
+if __name__ == '__main__': main()
+
+#-------------------------------------------------------------------------------
+# test funcs (to maybe relocate)
+
+def upload_test_file():
+    print("Start test upload to S3 ...")
+    s3 = make_s3_client()
+    print(s3)
+    fname = 'hemingway.txt'
+    resp = s3.upload_file(fname, S3_BUCKET, fname)
+    print(resp) 
+    print("SUCCESSFUL")
+
+def get_file():
+    key = 'output-1757986924786.json'
+    print('Grabbing file...') 
+    s3 = make_s3_client()
+    resp = s3.download_file(S3_BUCKET, key, '/tmp/'+key)
+    print(resp)
+    print('done grabbing file.')
+    #f = open('/tmp/'+key)
+    #print(f.readlines())

middleware/loglib.py

@@ -0,0 +1,24 @@
+import logging
+import sys
+
+AWS_TAG =   '[AWS] '
+SZ_TAG =    '[SZ] '
+DLQ_TAG =   '[DLQ] '
+
+_instantiated_loggers = {}
+
+def retrieve_logger(tag='default'):
+    global _instantiated_loggers
+    if tag in _instantiated_loggers:
+        return _instantiated_loggers[tag]
+    else:
+        x = logging.getLogger(tag)
+        x.setLevel(logging.INFO)
+        handler = logging.StreamHandler()
+        fmt = logging.Formatter(
+            '[%(asctime)s] [%(levelname)s] ' \
+            '[%(filename)s:%(lineno)s] %(message)s')
+        handler.setFormatter(fmt)
+        x.addHandler(handler)
+        _instantiated_loggers[tag] = x
+        return x

middleware/sample-data/hemingway.txt

@@ -0,0 +1,7 @@
+It was very late and everyone had left the cafe except an old man who sat in the
+shadow the leaves of the tree made against the electric light. In the day time the
+street was dusty, but at night the dew settled the dust and the old man liked to sit
+late because he was deaf and now at night it was quiet and he felt the difference.
+The two waiters inside the cafe knew that the old man was a little drunk, and while
+he was a good client they knew that if he became too drunk he would leave without
+paying, so they kept watch on him.