diff --git a/collectoss/application/cli/collection.py b/collectoss/application/cli/collection.py index 48a0090eb..251677854 100644 --- a/collectoss/application/cli/collection.py +++ b/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; diff --git a/collectoss/application/cli/github.py b/collectoss/application/cli/github.py index e3b8f75a2..98995458d 100644 --- a/collectoss/application/cli/github.py +++ b/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( """ diff --git a/collectoss/application/db/lib.py b/collectoss/application/db/lib.py index c5394365d..46d7d5926 100644 --- a/collectoss/application/db/lib.py +++ b/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 diff --git a/collectoss/tasks/data_analysis/insight_worker/tasks.py b/collectoss/tasks/data_analysis/insight_worker/tasks.py index 123407c58..134293b89 100644 --- a/collectoss/tasks/data_analysis/insight_worker/tasks.py +++ b/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)) diff --git a/collectoss/tasks/github/events.py b/collectoss/tasks/github/events.py index 24b1e42ff..ff14d4393 100644 --- a/collectoss/tasks/github/events.py +++ b/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""" diff --git a/collectoss/tasks/github/messages.py b/collectoss/tasks/github/messages.py index 342eeb2ca..c197ec069 100644 --- a/collectoss/tasks/github/messages.py +++ b/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"""