Optimize secondary task recollection

augur/application/db/lib.py

@@ -1,4 +1,4 @@
 import time
 import random
 import logging
 import sqlalchemy as s
@@ -9,7 +9,7 @@
 from psycopg2.errors import DeadlockDetected
 from typing import List, Any, Optional, Union
 
-from augur.application.db.models import Config, Repo, Commit, WorkerOauth, Issue, PullRequest, PullRequestReview, ContributorsAlias,    UnresolvedCommitEmail, Contributor, CollectionStatus
+from augur.application.db.models import Config, Repo, Commit, WorkerOauth, Issue, PullRequest, PullRequestReview, ContributorsAlias,UnresolvedCommitEmail, Contributor, CollectionStatus
 from augur.tasks.util.collection_state import CollectionState
 from augur.application.db import get_session, get_engine
 from augur.application.db.util import execute_session_query
@@ -17,12 +17,12 @@
 
 logger = logging.getLogger("db_lib")
 
 def convert_type_of_value(config_dict, logger=None):
 
 
     data_type = config_dict["type"]
 
     if data_type == "str" or data_type is None:
         return config_dict
 
     elif data_type == "int":
@@ -90,7 +90,7 @@
     with get_session() as session:
 
 
         # TODO temporary until added to the DB schema
         if section_name == "frontend" and setting_name == "pagination_offset":
             return 25
 
@@ -177,7 +177,7 @@
 
     try:
         working_commits = fetchall_data_from_sql_text(query)
     except:
         working_commits = []
 
     return working_commits
@@ -197,7 +197,7 @@
         return session.query(CollectionStatus).filter(getattr(CollectionStatus,f"{collection_type}_status" ) == CollectionState.COLLECTING.value).count()
 
 
 def facade_bulk_insert_commits(logger, records):
 
     with get_session() as session:
 
@@ -239,7 +239,7 @@
                 raise e
 
 
 def bulk_insert_dicts(logger, data: Union[List[dict], dict], table, natural_keys: List[str], return_columns: Optional[List[str]] = None, string_fields: Optional[List[str]] = None, on_conflict_update:bool = True) -> Optional[List[dict]]:
 
     if isinstance(data, list) is False:
 
@@ -509,4 +509,18 @@
                 WHERE issue_id = :issue_id
                 AND repo_id = :repo_id
             """)
-            connection.execute(update_stmt, update_data)
+            connection.execute(update_stmt, update_data)
+
+def get_secondary_data_last_collected(repo_id):
+
+    with get_session() as session:
+        try:
+           return session.query(CollectionStatus).filter(CollectionStatus.repo_id == repo_id).one().secondary_data_last_collected 
+        except s.orm.exc.NoResultFound:
+            return None
+
+def get_updated_prs(repo_id, since):
+
+    with get_session() as session:
+        return session.query(PullRequest).filter(PullRequest.repo_id == repo_id, PullRequest.pr_updated_at >= since).order_by(PullRequest.pr_src_number).all()
+

augur/tasks/data_analysis/__init__.py

@@ -1,7 +1,7 @@
 from celery import chain
 import logging 
 
-def machine_learning_phase(repo_git):
+def machine_learning_phase(repo_git, full_collection):
     from augur.tasks.data_analysis.clustering_worker.tasks import clustering_task
     from augur.tasks.data_analysis.discourse_analysis.tasks import discourse_analysis_task
     from augur.tasks.data_analysis.insight_worker.tasks import insight_task

augur/tasks/git/facade_tasks.py

@@ -410,7 +410,7 @@ def generate_contributor_sequence(logger,repo_git, session):
     return insert_facade_contributors.si(repo_id)
 
 
-def facade_phase(repo_git):
+def facade_phase(repo_git, full_collection):
     logger = logging.getLogger(facade_phase.__name__)
     logger.info("Generating facade sequence")
     facade_helper = FacadeHelper(logger)

augur/tasks/github/pull_requests/commits_model/core.py

@@ -2,20 +2,40 @@
 from augur.tasks.github.util.github_paginator import GithubPaginator
 from augur.application.db.models import *
 from augur.tasks.github.util.util import get_owner_repo
-from augur.application.db.lib import bulk_insert_dicts, fetchall_data_from_sql_text
+from augur.application.db.util import execute_session_query
+from augur.application.db.lib import get_secondary_data_last_collected, get_updated_prs
 
-def pull_request_commits_model(repo,logger, key_auth):
+
+def pull_request_commits_model(repo_id,logger, augur_db, key_auth, full_collection=False):
+
+    if full_collection:
+        # query existing PRs and the respective url we will append the commits url to
+        pr_url_sql = s.sql.text("""
+                SELECT DISTINCT pr_url, pull_requests.pull_request_id
+                FROM pull_requests--, pull_request_meta
+                WHERE repo_id = :repo_id
+            """).bindparams(repo_id=repo_id)
+        pr_urls = []
+        #pd.read_sql(pr_number_sql, self.db, params={})
+
+        pr_urls = augur_db.fetchall_data_from_sql_text(pr_url_sql)#session.execute_sql(pr_number_sql).fetchall()
 
-    # query existing PRs and the respective url we will append the commits url to
-    pr_url_sql = s.sql.text("""
-            SELECT DISTINCT pr_url, pull_requests.pull_request_id
-            FROM pull_requests--, pull_request_meta
-            WHERE repo_id = :repo_id
-        """).bindparams(repo_id=repo.repo_id)
-    pr_urls = []
-    #pd.read_sql(pr_number_sql, self.db, params={})
-
-    pr_urls = fetchall_data_from_sql_text(pr_url_sql)
+    else:
+        last_collected = get_secondary_data_last_collected(repo_id).date()
+        prs = get_updated_prs(repo_id, last_collected)
+        pr_urls = [pr.pr_url for pr in prs]
+
+        pr_urls = []
+        for pr in prs:
+            pr_urls.append({
+                'pr_url': pr.pr_url,
+                'pull_request_id': pr.pull_request_id
+            })
+
+
+    query = augur_db.session.query(Repo).filter(Repo.repo_id == repo_id)
+    repo = execute_session_query(query, 'one')
+
     owner, name = get_owner_repo(repo.repo_git)
 
     task_name = f"{owner}/{name} Pr commits"
@@ -53,6 +73,7 @@ def pull_request_commits_model(repo,logger, key_auth):
         logger.info(f"{task_name}: Inserting {len(all_data)} rows")
         pr_commits_natural_keys = ["pull_request_id", "repo_id", "pr_cmt_sha"]
         bulk_insert_dicts(logger, all_data,PullRequestCommit,pr_commits_natural_keys)
+        augur_db.insert_data(all_data,PullRequestCommit,pr_commits_natural_keys)
 
 
 

augur/tasks/github/pull_requests/commits_model/tasks.py

@@ -8,12 +8,12 @@
 
 
 @celery.task(base=AugurSecondaryRepoCollectionTask)
-def process_pull_request_commits(repo_git: str) -> None:
+def process_pull_request_commits(repo_git: str, full_collection: bool) -> None:
 
     logger = logging.getLogger(process_pull_request_commits.__name__)
 
     repo = get_repo_by_repo_git(repo_git)
 
     key_auth = GithubRandomKeyAuth(logger)
 
-    pull_request_commits_model(repo, logger, key_auth)
+    pull_request_commits_model(repo, logger, key_auth, full_collection)

augur/tasks/github/pull_requests/files_model/core.py

@@ -4,20 +4,35 @@
 from augur.tasks.github.util.util import get_owner_repo
 from augur.application.db.lib import bulk_insert_dicts, execute_sql
 from augur.application.db.util import execute_session_query
-import traceback 
-import httpx  # Import httpx
+from augur.application.db.lib import get_secondary_data_last_collected, get_updated_prs
 
-def pull_request_files_model(repo_id,logger, augur_db, key_auth):
-
-    # query existing PRs and the respective url we will append the commits url to
-    pr_number_sql = s.sql.text("""
-        SELECT DISTINCT pr_src_number as pr_src_number, pull_requests.pull_request_id
-        FROM pull_requests--, pull_request_meta
-        WHERE repo_id = :repo_id
-    """).bindparams(repo_id=repo_id)
-    pr_numbers = []
-    #pd.read_sql(pr_number_sql, self.db, params={})
+
+def pull_request_files_model(repo_id,logger, augur_db, key_auth, full_collection=False):
 
+    if full_collection:
+        # query existing PRs and the respective url we will append the commits url to
+        pr_number_sql = s.sql.text("""
+            SELECT DISTINCT pr_src_number as pr_src_number, pull_requests.pull_request_id
+            FROM pull_requests--, pull_request_meta
+            WHERE repo_id = :repo_id
+        """).bindparams(repo_id=repo_id)
+        pr_numbers = []
+        #pd.read_sql(pr_number_sql, self.db, params={})
+
+        result = augur_db.execute_sql(pr_number_sql)#.fetchall()
+        pr_numbers = [dict(row) for row in result.mappings()]
+
+    else:
+        last_collected = get_secondary_data_last_collected(repo_id).date()
+        prs = get_updated_prs(repo_id, last_collected)
+
+        pr_numbers = []
+        for pr in prs:
+            pr_numbers.append({
+                'pr_src_number': pr.pr_src_number,
+                'pull_request_id': pr.pull_request_id
+            })
+
     query = augur_db.session.query(Repo).filter(Repo.repo_id == repo_id)
     repo = execute_session_query(query, 'one')
     owner, name = get_owner_repo(repo.repo_git)
@@ -78,4 +93,4 @@ def pull_request_files_model(repo_id,logger, augur_db, key_auth):
     if len(pr_file_rows) > 0:
         # Execute a bulk upsert with sqlalchemy 
         pr_file_natural_keys = ["pull_request_id", "repo_id", "pr_file_path"]
-        bulk_insert_dicts(logger, pr_file_rows, PullRequestFile, pr_file_natural_keys)
+        augur_db.insert_data(pr_file_rows, PullRequestFile, pr_file_natural_keys)

augur/tasks/github/pull_requests/files_model/tasks.py

@@ -6,7 +6,7 @@
 from augur.application.db.util import execute_session_query
 
 @celery.task(base=AugurSecondaryRepoCollectionTask)
-def process_pull_request_files(repo_git: str) -> None:
+def process_pull_request_files(repo_git: str, full_collection: bool) -> None:
 
     logger = logging.getLogger(process_pull_request_files.__name__)
 
@@ -15,4 +15,4 @@ def process_pull_request_files(repo_git: str) -> None:
         query = augur_db.session.query(Repo).filter(Repo.repo_git == repo_git)
         repo = execute_session_query(query, 'one')
 
-        pull_request_files_model(repo.repo_id, logger, augur_db, manifest.key_auth)
+        pull_request_files_model(repo.repo_id, logger, augur_db, manifest.key_auth, full_collection)

augur/tasks/github/pull_requests/tasks.py

@@ -12,9 +12,7 @@
 from augur.application.db.lib import get_session, get_repo_by_repo_git, bulk_insert_dicts, get_pull_request_reviews_by_repo_id
 from augur.application.db.util import execute_session_query
 from ..messages.tasks import process_github_comment_contributors
-from augur.tasks.github.util.github_random_key_auth import GithubRandomKeyAuth
-
-import httpx
+from augur.application.db.lib import get_secondary_data_last_collected, get_updated_prs
 
 from typing import Generator, List, Dict
 
@@ -328,7 +326,7 @@ def collect_pull_request_review_comments(repo_git: str) -> None:
 
 
 @celery.task(base=AugurSecondaryRepoCollectionTask)
-def collect_pull_request_reviews(repo_git: str) -> None:
+def collect_pull_request_reviews(repo_git: str, full_collection: bool) -> None:
 
     logger = logging.getLogger(collect_pull_request_reviews.__name__)
 
@@ -339,13 +337,20 @@ def collect_pull_request_reviews(repo_git: str) -> None:
     data_source = "Github API"
 
     repo_id = get_repo_by_repo_git(repo_git).repo_id
+    with GithubTaskManifest(logger) as manifest:
 
-    key_auth = GithubRandomKeyAuth(logger)
+        augur_db = manifest.augur_db
 
-    with get_session() as session:
-
-        query = session.query(PullRequest).filter(PullRequest.repo_id == repo_id).order_by(PullRequest.pr_src_number)
-        prs = execute_session_query(query, 'all')
+        query = augur_db.session.query(Repo).filter(Repo.repo_git == repo_git)
+        repo_id = execute_session_query(query, 'one').repo_id
+
+        if full_collection:
+
+            query = augur_db.session.query(PullRequest).filter(PullRequest.repo_id == repo_id).order_by(PullRequest.pr_src_number)
+            prs = execute_session_query(query, 'all')
+        else:
+            last_collected = get_secondary_data_last_collected(repo_id).date()
+            prs = get_updated_prs(repo_id, last_collected)
 
     pr_count = len(prs)
 
@@ -398,8 +403,8 @@ def collect_pull_request_reviews(repo_git: str) -> None:
             if contributor:
                 contributors.append(contributor)
 
-    logger.info(f"{owner}/{repo} Pr reviews: Inserting {len(contributors)} contributors")
-    bulk_insert_dicts(logger, contributors, Contributor, ["cntrb_id"])
+        logger.info(f"{owner}/{repo} Pr reviews: Inserting {len(contributors)} contributors")
+        augur_db.insert_data(contributors, Contributor, ["cntrb_id"])
 
 
     pr_reviews = []
@@ -411,9 +416,9 @@ def collect_pull_request_reviews(repo_git: str) -> None:
             if "cntrb_id" in review:
                 pr_reviews.append(extract_needed_pr_review_data(review, pull_request_id, repo_id, platform_id, tool_source, tool_version))
 
-    logger.info(f"{owner}/{repo}: Inserting pr reviews of length: {len(pr_reviews)}")
-    pr_review_natural_keys = ["pr_review_src_id",]
-    bulk_insert_dicts(logger, pr_reviews, PullRequestReview, pr_review_natural_keys)
+        logger.info(f"{owner}/{repo}: Inserting pr reviews of length: {len(pr_reviews)}")
+        pr_review_natural_keys = ["pr_review_src_id",]
+        augur_db.insert_data(pr_reviews, PullRequestReview, pr_review_natural_keys)
 
 
 

augur/tasks/start_tasks.py

@@ -43,21 +43,21 @@
 """
 
 #Prelim phases are used to detect if where the repo has hosted has moved or not.
-def prelim_phase(repo_git):
+def prelim_phase(repo_git, full_collection):
 
     logger = logging.getLogger(prelim_phase.__name__)
 
     return detect_github_repo_move_core.si(repo_git)
 
-def prelim_phase_secondary(repo_git):
+def prelim_phase_secondary(repo_git, full_collection):
     logger = logging.getLogger(prelim_phase.__name__)
 
     return detect_github_repo_move_secondary.si(repo_git)
 
 
 #This is the phase that defines the message for core augur collection
 #A chain is needed for each repo.
-def primary_repo_collect_phase(repo_git):
+def primary_repo_collect_phase(repo_git, full_collection):
     logger = logging.getLogger(primary_repo_collect_phase.__name__)
 
 
@@ -86,7 +86,7 @@ def primary_repo_collect_phase(repo_git):
 
     return repo_task_group
 
-def primary_repo_collect_phase_gitlab(repo_git):
+def primary_repo_collect_phase_gitlab(repo_git, full_collection):
 
     logger = logging.getLogger(primary_repo_collect_phase_gitlab.__name__)
 
@@ -110,13 +110,13 @@ def primary_repo_collect_phase_gitlab(repo_git):
 
 #This phase creates the message for secondary collection tasks.
 #These are less important and have their own worker.
-def secondary_repo_collect_phase(repo_git):
+def secondary_repo_collect_phase(repo_git, full_collection):
     logger = logging.getLogger(secondary_repo_collect_phase.__name__)
 
     repo_task_group = group(
-        process_pull_request_files.si(repo_git),
-        process_pull_request_commits.si(repo_git),
-        chain(collect_pull_request_reviews.si(repo_git), collect_pull_request_review_comments.si(repo_git)),
+        process_pull_request_files.si(repo_git, full_collection),
+        process_pull_request_commits.si(repo_git, full_collection),
+        chain(collect_pull_request_reviews.si(repo_git, full_collection), collect_pull_request_review_comments.si(repo_git)),
         process_ossf_dependency_metrics.si(repo_git)
     )
 
@@ -167,7 +167,7 @@ def build_primary_repo_collect_request(session, logger, enabled_phase_names, day
     primary_gitlab_enabled_phases.append(primary_repo_collect_phase_gitlab)
 
     #task success is scheduled no matter what the config says.
-    def core_task_success_util_gen(repo_git):
+    def core_task_success_util_gen(repo_git, full_collection):
         return core_task_success_util.si(repo_git)
 
     primary_enabled_phases.append(core_task_success_util_gen)
@@ -187,7 +187,7 @@ def build_secondary_repo_collect_request(session, logger, enabled_phase_names, d
 
     secondary_enabled_phases.append(secondary_repo_collect_phase)
 
-    def secondary_task_success_util_gen(repo_git):
+    def secondary_task_success_util_gen(repo_git, full_collection):
         return secondary_task_success_util.si(repo_git)
 
     secondary_enabled_phases.append(secondary_task_success_util_gen)
@@ -203,12 +203,12 @@ def build_facade_repo_collect_request(session, logger, enabled_phase_names, days
 
     facade_enabled_phases.append(facade_phase)
 
-    def facade_task_success_util_gen(repo_git):
+    def facade_task_success_util_gen(repo_git, full_collection):
         return facade_task_success_util.si(repo_git)
 
     facade_enabled_phases.append(facade_task_success_util_gen)
 
-    def facade_task_update_weight_util_gen(repo_git):
+    def facade_task_update_weight_util_gen(repo_git, full_collection):
         return git_update_commit_count_weight.si(repo_git)
 
     facade_enabled_phases.append(facade_task_update_weight_util_gen)
@@ -223,7 +223,7 @@ def build_ml_repo_collect_request(session, logger, enabled_phase_names, days_unt
 
     ml_enabled_phases.append(machine_learning_phase)
 
-    def ml_task_success_util_gen(repo_git):
+    def ml_task_success_util_gen(repo_git, full_collection):
         return ml_task_success_util.si(repo_git)
 
     ml_enabled_phases.append(ml_task_success_util_gen)