added in batch inserts

algo.py

@@ -14,17 +14,10 @@ from db.db_utils import store_mapping_to_db, cached_get_mapping_from_db, get_dic
 from ask_gpt import query_gpt
 from multi_food_item_detector import extract_items, has_delimiters
 from mapping_template import empty_template, heterogeneous_template, multi_item_template, nonfood_template, usda_template
-# from tasks import insert_result
 from specificity_classifier import classify_text_to_specificity
 
-logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s')
 similarity_threshold = 0.78
-
-
-def insert_result(db_conn, run_key, mappings):
-    db_cursor = db_conn.cursor()
-    for mapping in mappings:
-        store_result_to_db(db_cursor, db_conn, run_key, mapping)
+logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s')
 
 
 class Algo:
@@ -347,12 +340,12 @@ class Algo:
                 # store_result_to_db(self.db_cursor, self.db_conn, self.run_key, mapping)
                 results.append(mapping)
 
-            if len(result_batch) >= 100:
-                insert_result(self.db_conn, self.run_key, result_batch)
+            if len(result_batch) >= 500:
+                store_batch_results_to_db(self.db_conn, self.run_key, result_batch)
                 result_batch = []
 
         if len(result_batch) > 0:
-            insert_result(self.db_conn, self.run_key, result_batch)
+            store_batch_results_to_db(self.db_conn, self.run_key, result_batch)
             result_batch = []
 
 

db/db_utils.py

@@ -3,9 +3,10 @@ import psycopg2
 import logging
 from dotenv import load_dotenv
 from functools import lru_cache
+from psycopg2.extras import execute_values
 
-load_dotenv()
 
+load_dotenv()
 
 def get_connection():
     DATABASE_URL = os.environ['DATABASE_URL']
@@ -19,7 +20,6 @@ def get_connection():
         print(f"Failed to connect to database: {e}")
         raise
 
-
 def initialize_db(conn):
     cursor = conn.cursor()
     cursor.execute('''
@@ -215,3 +215,69 @@ def store_result_to_db(cursor, conn, run_key, result):
     
     conn.commit()
     return True
+
+
+def store_batch_results_to_db(cursor, conn, run_key, results):
+    values = [
+        (
+            run_key,
+            result['run_row'],
+            result['date'], 
+            result['input_word'], 
+            result['dictionary_word'], 
+            result['is_food'], 
+            result['sr_legacy_food_category'], 
+            result['wweia_category'], 
+            result['dry_matter_content'],
+            result['leakage'],
+            result['weight'],
+            result['weight_metric_tonnes'],
+            result['donor'],
+            result['similarity_score'], 
+            result['food_nonfood_score'],
+            result['distance'],
+            result['ef'], 
+            result['mt_lb_mile'], 
+            result['baseline_emissions'], 
+            result['leakage_emissions'],
+            result['project_emissions'],  
+            result['total_emissions_reduction']
+        )
+        for result in results
+    ]
+    
+    insert_query = '''
+        INSERT INTO results (
+            run_key, run_row, date, input_word, dictionary_word, is_food, 
+            sr_legacy_food_category, wweia_category, dry_matter_content, leakage, 
+            weight, weight_metric_tonnes, donor, similarity_score, food_nonfood_score, 
+            distance, ef, mt_lb_mile, baseline_emissions, leakage_emissions, 
+            project_emissions, total_emissions_reduction
+        ) VALUES %s
+        ON CONFLICT (run_key, run_row) 
+        DO UPDATE SET 
+            date = EXCLUDED.date,
+            input_word = EXCLUDED.input_word,
+            dictionary_word = EXCLUDED.dictionary_word,
+            is_food = EXCLUDED.is_food,
+            sr_legacy_food_category = EXCLUDED.sr_legacy_food_category,
+            wweia_category = EXCLUDED.wweia_category,
+            dry_matter_content = EXCLUDED.dry_matter_content,
+            leakage = EXCLUDED.leakage,
+            weight = EXCLUDED.weight,
+            weight_metric_tonnes = EXCLUDED.weight_metric_tonnes,
+            donor = EXCLUDED.donor,
+            similarity_score = EXCLUDED.similarity_score,
+            food_nonfood_score = EXCLUDED.food_nonfood_score,
+            distance = EXCLUDED.distance,
+            ef = EXCLUDED.ef,
+            mt_lb_mile = EXCLUDED.mt_lb_mile,
+            baseline_emissions = EXCLUDED.baseline_emissions,
+            leakage_emissions = EXCLUDED.leakage_emissions,
+            project_emissions = EXCLUDED.project_emissions,
+            total_emissions_reduction = EXCLUDED.total_emissions_reduction;
+    '''
+    
+    execute_values(cursor, insert_query, values)
+    conn.commit()
+    return True

tasks.py

@@ -7,30 +7,11 @@ from algo import Algo
 from dotenv import load_dotenv
 from redis import Redis
 from rq import Queue
-# from celery import Celery
 from db.db_utils import get_connection, store_result_to_db
 
 load_dotenv()
 
-# app = Celery('tasks', broker=REDIS_URL, backend=REDIS_URL)
 
-# app.conf.update(
-#     result_expires=3600,
-#     task_serializer='json',
-#     result_serializer='json',
-#     accept_content=['json'],
-#     timezone='UTC',
-#     enable_utc=True,
-#     broker_connection_retry_on_startup=True
-# )
-
-# @app.task
-# def insert_result(db_conn, run_key, mappings):
-#     db_cursor = db_conn.cursor()
-#     for mapping in mappings:
-#         store_result_to_db(db_cursor, db_conn, run_key, mapping)
-
-# @app.task
 def process_file(raw_file_name):
     print(f"Processing {raw_file_name}")
     if not raw_file_name.endswith('.csv'):