[tasks/github] Fix uncommitted UPDATE in update_issue_closed_cntrbs_by_repo_id

collectoss/application/cli/collection.py

@@ -201,7 +201,7 @@ def repo_reset(ctx):
     """
     Refresh repo collection to force data collection
     """
-    with ctx.obj.engine.connect() as connection:
+    with ctx.obj.engine.begin() as connection:
         connection.execute(s.sql.text("""
             UPDATE operations.collection_status 
             SET core_status='Pending',core_task_id = NULL, core_data_last_collected = NULL;

collectoss/application/cli/github.py

@@ -26,7 +26,7 @@ def update_api_key():
     Get the ratelimit of Github API keys
     """
 
-    with DatabaseEngine() as engine, engine.connect() as connection:
+    with DatabaseEngine() as engine, engine.begin() as connection:
 
         get_api_keys_sql = s.sql.text(
             """

collectoss/application/db/lib.py

@@ -546,7 +546,10 @@ def update_issue_closed_cntrbs_by_repo_id(repo_id):
         )
 
     if update_data:
-        with engine.connect() as connection:
+        # engine.begin() auto-commits (or rolls back on exception).
+        # engine.connect() does NOT auto-commit, so the UPDATE was previously
+        # silently rolled back on every call (issue #3457).
+        with engine.begin() as connection:
             update_stmt = s.text("""
                 UPDATE issues
                 SET cntrb_id = :cntrb_id

collectoss/tasks/data_analysis/insight_worker/tasks.py

@@ -128,7 +128,7 @@ def insight_model(repo_git: str,logger,engine) -> None:
             AND repo_insights.ri_field = to_delete.ri_field
     """)
 
-    with engine.connect() as conn:
+    with engine.begin() as conn:
         result = conn.execute(delete_points_SQL, parameters=dict(repo_id=repo_id, min_date=min_date))
 
     # get table values to check for dupes later on
@@ -561,7 +561,7 @@ def clear_insights(repo_id, new_endpoint, new_field, logger):
                 AND ri_field = '{}'
     """.format(repo_id, new_endpoint, new_field)
     try:
-        with engine.connect() as conn:
+        with engine.begin() as conn:
             result = conn.execute(deleteSQL)
     except Exception as e:
         logger.info("Error occured deleting insight slot: {}".format(e))
@@ -579,7 +579,7 @@ def clear_insights(repo_id, new_endpoint, new_field, logger):
                 AND ri_field = '{}'
     """.format(repo_id, new_endpoint, new_field)
     try:
-        with engine.connect() as conn:
+        with engine.begin() as conn:
             result = conn.execute(deleteSQL)
     except Exception as e:
         logger.info("Error occured deleting insight slot: {}".format(e))
@@ -622,7 +622,7 @@ def clear_insight(repo_id, new_score, new_metric, new_field, logger):
                             AND ri_field = '{}'
                 """.format(record['repo_id'], record['ri_metric'], record['ri_field'])
                 try:
-                    with engine.connect() as conn:
+                    with engine.begin() as conn:
                         result = conn.execute(deleteSQL)
                 except Exception as e:
                     logger.info("Error occured deleting insight slot: {}".format(e))
@@ -676,7 +676,7 @@ def clear_insight(repo_id, new_score, new_metric, new_field, logger):
                     AND ri_metric = '{}'
         """.format(insight['repo_id'], insight['ri_metric'])
         try:
-            with engine.connect() as conn:
+            with engine.begin() as conn:
                 result = conn.execute(deleteSQL)
         except Exception as e:
             logger.info("Error occured deleting insight slot: {}".format(e))

collectoss/tasks/github/events.py

@@ -167,7 +167,10 @@ def _process_events(self, events, repo_id):
         self._process_issue_events(issue_events, repo_id)
         self._process_pr_events(pr_events, repo_id)
 
-        update_issue_closed_cntrbs_by_repo_id(repo_id)
+        try:
+            update_issue_closed_cntrbs_by_repo_id(repo_id)
+        except Exception as e:
+            self._logger.error(f"{self.repo_identifier} - {self.task_name}: Failed to update issue closed contributors: {e}")
 
     def _process_issue_events(self, issue_events, repo_id):
 
@@ -284,7 +287,7 @@ def _collect_and_process_issue_events(self, owner, repo, repo_id, key_auth, sinc
 
         engine = get_engine()
 
-        with engine.connect() as connection:
+        with engine.begin() as connection:
 
             if since:
                 # TODO: Remove src id if it ends up not being needed
@@ -349,7 +352,7 @@ def _collect_and_process_pr_events(self, owner, repo, repo_id, key_auth, since):
 
         engine = get_engine()
 
-        with engine.connect() as connection:
+        with engine.begin() as connection:
 
             if since:
                 query = text(f"""

collectoss/tasks/github/messages.py

@@ -93,7 +93,7 @@ def process_large_issue_and_pr_message_collection(repo_id, repo_git: str, logger
 
     engine = get_engine()
 
-    with engine.connect() as connection:
+    with engine.begin() as connection:
 
         if since:
              query = text(f"""