Fix Facade Task Signature to Take repo_git instead of repo_id

augur/tasks/git/facade_tasks.py

@@ -1,4 +1,4 @@
 #SPDX-License-Identifier: MIT
 
 import logging
 from celery import group, chain
@@ -13,7 +13,7 @@
 from augur.tasks.git.util.facade_worker.facade_worker.postanalysiscleanup import git_repo_cleanup
 
 
 from augur.tasks.github.facade_github.tasks import *
 from augur.tasks.util.collection_state import CollectionState
 from augur.tasks.util.collection_util import get_collection_status_repo_git_from_filter
 from augur.tasks.git.util.facade_worker.facade_worker.repofetch import GitCloneError, git_repo_initialize, git_repo_updates
@@ -30,13 +30,13 @@
 from augur.tasks.git.dependency_libyear_tasks.tasks import process_libyear_dependency_metrics
 from augur.tasks.git.scc_value_tasks.tasks import process_scc_value_metrics
 
 from augur.tasks.github.util.github_task_session import *
 
 
 #define an error callback for chains in facade collection so facade doesn't make the program crash
 #if it does.
 @celery.task
 def facade_error_handler(request,exc,traceback):
 
     logger = logging.getLogger(facade_error_handler.__name__)
 
@@ -61,10 +61,10 @@
     with FacadeSession(logger) as session:
 
         repo = session.query(Repo).filter(Repo.repo_git == repo_git).one()
         repo_id = repo.repo_id
 
         session.update_status('Running analysis')
         session.log_activity('Info',f"Beginning analysis.")
 
 
 @celery.task(base=AugurFacadeRepoCollectionTask)
@@ -86,7 +86,7 @@
 
             try:
                 session.execute_sql(log_message)
             except:
                 pass
 
 
@@ -140,7 +140,7 @@
 
         #Get the huge list of commits to process.
         absoulte_path = get_absolute_repo_path(session.repo_base_directory, repo.repo_id, repo.repo_path,repo.repo_name)
         repo_loc = (f"{absoulte_path}/.git")
         # Grab the parents of HEAD
 
         parent_commits = get_parent_commits_set(repo_loc, start_date)
@@ -216,7 +216,7 @@
 
         #Get the huge list of commits to process.
         absoulte_path = get_absolute_repo_path(session.repo_base_directory, repo.repo_id, repo.repo_path, repo.repo_name)
         repo_loc = (f"{absoulte_path}/.git")
         # Grab the parents of HEAD
 
         parent_commits = get_parent_commits_set(repo_loc, start_date)
@@ -238,7 +238,7 @@
 
         logger.info(f"Got to analysis!")
         absoulte_path = get_absolute_repo_path(session.repo_base_directory, repo.repo_id, repo.repo_path,repo.repo_name)
         repo_loc = (f"{absoulte_path}/.git")
 
         pendingCommitRecordsToInsert = []
 
@@ -439,32 +439,6 @@
     return analysis_sequence
 
 
-
-def generate_contributor_sequence(logger,repo_git, session):
-
-    contributor_sequence = []
-    #all_repo_ids = []
-    repo_id = None
-
-    #contributor_sequence.append(facade_start_contrib_analysis_task.si())
-    query = s.sql.text("""SELECT repo_id FROM repo
-    WHERE repo_git=:value""").bindparams(value=repo_git)
-
-    repo = session.execute_sql(query).fetchone()
-    session.logger.info(f"repo: {repo}")
-    repo_id = repo[0]
-    #pdb.set_trace()
-    #breakpoint()
-    #for repo in all_repos:
-    #    contributor_sequence.append(insert_facade_contributors.si(repo['repo_id']))
-    #all_repo_ids = [repo['repo_id'] for repo in all_repos]
-
-    #contrib_group = create_grouped_task_load(dataList=all_repo_ids,task=insert_facade_contributors)#group(contributor_sequence)
-    #contrib_group.link_error(facade_error_handler.s())
-    #return contrib_group#chain(facade_start_contrib_analysis_task.si(), contrib_group)
-    return insert_facade_contributors.si(repo_id)
-
-
 def facade_phase(repo_git):
     logger = logging.getLogger(facade_phase.__name__)
     logger.info("Generating facade sequence")
@@ -506,7 +480,7 @@
 
         #Generate contributor analysis task group.
         if not limited_run or (limited_run and run_facade_contributors):
-            facade_core_collection.append(generate_contributor_sequence(logger,repo_git,session))
+            facade_core_collection.append(insert_facade_contributors.si(repo_git))
 
 
         #These tasks need repos to be cloned by facade before they can work.

augur/tasks/github/facade_github/tasks.py

@@ -194,14 +194,22 @@ def link_commits_to_contributor(session,contributorQueue):
 
 # Update the contributors table from the data facade has gathered.
 @celery.task(base=AugurFacadeRepoCollectionTask, bind=True)
-def insert_facade_contributors(self, repo_id):
+def insert_facade_contributors(self, repo_git):
 
     engine = self.app.engine
 
     logger = logging.getLogger(insert_facade_contributors.__name__)
+    repo_id = None
 
     with GithubTaskManifest(logger) as manifest:
 
+        #contributor_sequence.append(facade_start_contrib_analysis_task.si())
+        query = s.sql.text("""SELECT repo_id FROM repo
+        WHERE repo_git=:value""").bindparams(value=repo_git)
+
+        repo =  manifest.augur_db.execute_sql(query).fetchone()
+        logger.info(f"repo: {repo}")
+        repo_id = repo[0]
 
         # Get all of the commit data's emails and names from the commit table that do not appear
         # in the contributors table or the contributors_aliases table.

augur/tasks/start_tasks.py

@@ -326,13 +326,22 @@ def retry_errored_repos(self):
     #collection_status table once augur dev is less unstable.
     with DatabaseSession(logger,engine) as session:
         query = s.sql.text(f"""UPDATE collection_status SET secondary_status = '{CollectionState.PENDING.value}'"""
-        f""" WHERE secondary_status = '{CollectionState.ERROR.value}' ;"""
+        f""" WHERE secondary_status = '{CollectionState.ERROR.value}' and secondary_data_last_collected is NULL;"""
         f"""UPDATE collection_status SET core_status = '{CollectionState.PENDING.value}'"""
-        f""" WHERE core_status = '{CollectionState.ERROR.value}' ;"""
+        f""" WHERE core_status = '{CollectionState.ERROR.value}' and core_data_last_collected is NULL;"""
         f"""UPDATE collection_status SET facade_status = '{CollectionState.PENDING.value}'"""
-        f""" WHERE facade_status = '{CollectionState.ERROR.value}' ;"""
+        f""" WHERE facade_status = '{CollectionState.ERROR.value}' and facade_data_last_collected is NULL;"""
         f"""UPDATE collection_status SET ml_status = '{CollectionState.PENDING.value}'"""
-        f""" WHERE ml_status = '{CollectionState.ERROR.value}' ;"""
+        f""" WHERE ml_status = '{CollectionState.ERROR.value}' and ml_data_last_collected is NULL;"""
+
+        f"""UPDATE collection_status SET secondary_status = '{CollectionState.SUCCESS.value}'"""
+        f""" WHERE secondary_status = '{CollectionState.ERROR.value}' and secondary_data_last_collected is not NULL;"""
+        f"""UPDATE collection_status SET core_status = '{CollectionState.SUCCESS.value}'"""
+        f""" WHERE core_status = '{CollectionState.ERROR.value}' and core_data_last_collected is not NULL;;"""
+        f"""UPDATE collection_status SET facade_status = '{CollectionState.SUCCESS.value}'"""
+        f""" WHERE facade_status = '{CollectionState.ERROR.value}' and facade_data_last_collected is not NULL;;"""
+        f"""UPDATE collection_status SET ml_status = '{CollectionState.SUCCESS.value}'"""
+        f""" WHERE ml_status = '{CollectionState.ERROR.value}' and ml_data_last_collected is not NULL;;"""
         )
 
         session.execute_sql(query)