make changes for integration

Author: saquibb8

…ng/src/saayam-org-aggregator/__init__.py …/src/aggregate-daily-metrics/__init__.pydata-engineering/src/saayam-org-aggregator/__init__.py renamed to data-engineering/src/aggregate-daily-metrics/__init__.py data-engineering/src/saayam-org-aggregator/__init__.py renamed to data-engineering/src/aggregate-daily-metrics/__init__.py

@@ -1,43 +1,39 @@
-import subprocess
 import boto3
-import sys
 import json
 from aws_lambda_powertools.utilities import parameters
 import pandas as pd
-
-subprocess.call([sys.executable, "-m", "pip", "install", "pg8000", "-t", "/tmp/"])
-sys.path.insert(0, "/tmp/")
 import pg8000
 
+
 GEN_AI_LAMBDA = "More_Org_GenAI_Py_v3126"
 
+# --- All cached at module level, initialized once on cold start ---
+lambda_client = boto3.client('lambda')
+
+_creds = json.loads(parameters.get_parameter(
+    '/dev/saayam/db/Virginia/Analytics/user',
+    decrypt=True,
+    max_age=3600
+))
+_db_name = _creds['DATABASE NAME']
+_db_conn = pg8000.connect(
+    host=_creds['HOST'],
+    user=_creds['USERNAME'],
+    password=_creds['PASSWORD'],
+    database=_db_name,
+    port=_creds['PORT'],
+    ssl_context=True
+)
+# -----------------------------------------------------------------
+
 def get_orgs_from_db(location, category):
     try:
-        creds = json.loads(parameters.get_parameter(
-            '/dev/saayam/db/Virginia/Analytics/user',
-            decrypt=True,
-            max_age=3600
-        ))
-        database = creds['DATABASE NAME']
-
-        conn = pg8000.connect(
-            host=creds['HOST'],
-            user=creds['USERNAME'],
-            password=creds['PASSWORD'],
-            database=database,
-            port=creds['PORT'],
-            ssl_context=True
-        )
-
         df = pd.read_sql(
-            f"SELECT * FROM {database}.organizations WHERE mission = '{category}' AND city_name = '{location}'",
-            conn
+            f"SELECT * FROM {_db_name}.organizations WHERE mission = '{category}' AND city_name = '{location}'",
+            _db_conn
         )
-        conn.close()
+        df["db_or_ai"] = "db"
         return df
-
-    except parameters.GetParameterError as e:
-        raise Exception(f'Failed to retrieve DB credentials: {str(e)}')
     except pg8000.DatabaseError as e:
         raise Exception(f'Database error: {str(e)}')
     except Exception as e:
@@ -46,7 +42,7 @@ def get_orgs_from_db(location, category):
 
 def get_ai_orgs(subject, description, location):
     try:
-        response = boto3.client('lambda').invoke(
+        response = lambda_client.invoke(
             FunctionName=GEN_AI_LAMBDA,
             InvocationType='RequestResponse',
             Payload=json.dumps({
@@ -55,14 +51,12 @@ def get_ai_orgs(subject, description, location):
                 "location": location
             })
         )
-
         payload = json.loads(response['Payload'].read())
-
         if payload.get('statusCode') != 200:
             raise Exception(f'GenAI Lambda returned error: {payload}')
-
-        return pd.DataFrame(payload['body']['organizations'])
-
+        orgs = pd.DataFrame(payload['body']['organizations'])
+        orgs["db_or_ai"] = "ai"
+        return orgs
     except boto3.exceptions.Boto3Error as e:
         raise Exception(f'Failed to invoke GenAI Lambda: {str(e)}')
     except (KeyError, TypeError) as e:
@@ -77,14 +71,13 @@ def merge_organizations(db_organizations, genAI_organizations):
             'org_name': 'name',
             'city_name': 'location',
             'phone': 'contact'
-        })[['name', 'location', 'contact', 'email', 'web_url', 'mission', 'source']]
+        })[['name', 'location', 'contact', 'email', 'web_url', 'mission', 'source', "db_or_ai"]]
 
         genAI_organizations = genAI_organizations.rename(columns={
             'organization_name': 'name'
-        })[['name', 'location', 'contact', 'email', 'web_url', 'mission', 'source']]
+        })[['name', 'location', 'contact', 'email', 'web_url', 'mission', 'source', "db_or_ai"]]
 
         return pd.concat([db_organizations, genAI_organizations], ignore_index=True)
-
     except KeyError as e:
         raise Exception(f'Missing expected column during merge: {str(e)}')
     except Exception as e:

data-engineering/src/aggregate-daily-metrics/helpers.py

@@ -1,8 +1,8 @@
 import json
 from helpers import get_ai_orgs, get_orgs_from_db, merge_organizations
+from concurrent.futures import ThreadPoolExecutor
 
-VERSION = "1.0.1"  # Test auto-deploy
-
+# handle the lambda function call
 def lambda_handler(event, context):
     try:
         raw_body = event.get("body")
@@ -19,13 +19,27 @@ def lambda_handler(event, context):
                 'body': json.dumps({'error': 'location and category are required fields'})
             }
 
-        db_organizations = get_orgs_from_db(location, category)
-        genAI_organizations = get_ai_orgs(subject, description, location)
+        # db_organizations = get_orgs_from_db(location, category)
+        # genAI_organizations = get_ai_orgs(subject, description, location)
+
+        #Implemented parallelization to speed up the reponse
+
+        with ThreadPoolExecutor() as executor:
+            db_future = executor.submit(get_orgs_from_db, location, category)
+            ai_future = executor.submit(get_ai_orgs, subject, description, location)
+
+            db_organizations = db_future.result()
+            genAI_organizations = ai_future.result()
+
         combined_list = merge_organizations(db_organizations, genAI_organizations)
 
         return {
             'statusCode': 200,
-            'body': combined_list.to_dict(orient='records')
+            'headers': {
+                "Content-Type": "application/json",
+                "Access-Control-Allow-Origin": '*'
+            },
+            'body': json.dumps(combined_list.to_dict(orient='records'))
         }
 
     except json.JSONDecodeError as e: