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"
@@ -52,7 +72,7 @@ def pull_request_commits_model(repo,logger, key_auth):
     if len(all_data) > 0:
         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

@@ -2,18 +2,18 @@
 from augur.tasks.github.pull_requests.commits_model.core import *
 from augur.tasks.init.celery_app import celery_app as celery
 from augur.tasks.init.celery_app import AugurSecondaryRepoCollectionTask
-from augur.tasks.github.util.github_random_key_auth import GithubRandomKeyAuth
+from augur.tasks.github.util.github_task_session import GithubTaskManifest
 from augur.application.db.lib import get_repo_by_repo_git
 
 
 
 @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)
+    with GithubTaskManifest(logger) as manifest:
 
-    pull_request_commits_model(repo, logger, key_auth)
+        pull_request_commits_model(repo.repo_id, logger, manifest.augur_db, manifest.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

@@ -9,12 +9,11 @@
 from augur.tasks.github.util.util import add_key_value_pair_to_dicts, get_owner_repo
 from augur.application.db.models import PullRequest, Message, PullRequestReview, PullRequestLabel, PullRequestReviewer, PullRequestMeta, PullRequestAssignee, PullRequestReviewMessageRef, Contributor, Repo
 from augur.tasks.github.util.github_task_session import GithubTaskManifest
+from augur.tasks.github.util.github_random_key_auth import GithubRandomKeyAuth
 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 +327,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,81 +338,88 @@ 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
 
-    pr_count = len(prs)
+        if full_collection:
 
-    all_pr_reviews = {}
-    for index, pr in enumerate(prs):
+            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_number = pr.pr_src_number
-        pull_request_id = pr.pull_request_id
+        pr_count = len(prs)
 
-        logger.info(f"{owner}/{repo} Collecting Pr Reviews for pr {index + 1} of {pr_count}")
+        all_pr_reviews = {}
+        for index, pr in enumerate(prs):
 
-        pr_review_url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
+            pr_number = pr.pr_src_number
+            pull_request_id = pr.pull_request_id
 
-        pr_reviews = []
-        pr_reviews_generator = GithubPaginator(pr_review_url, key_auth, logger)
-        for page_data, page in pr_reviews_generator.iter_pages():
-
-            if page_data is None:
-                break
-
-            if len(page_data) == 0:
-                break
-
-            if isinstance(page_data, list):
-                page_data = [
-                    element.decode('utf-8').replace('\x00', ' ') if isinstance(element, bytes) else element
-                    for element in page_data
-                ]
-                logger.info(f"NUL characters were found in PR Reviews and replaced with spaces.") 
-            elif isinstance(page_data, bytes):
-                page_data = page_data.decode('utf-8').replace('\x00', ' ')
-                logger.info(f"NUL characters were found in PR Reviews and replaced with spaces.") 
+            logger.info(f"{owner}/{repo} Collecting Pr Reviews for pr {index + 1} of {pr_count}")
+
+            pr_review_url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
+
+            pr_reviews = []
+            pr_reviews_generator = GithubPaginator(pr_review_url, manifest.key_auth, logger)
+            for page_data, page in pr_reviews_generator.iter_pages():
+
+                if page_data is None:
+                    break
+
+                if len(page_data) == 0:
+                    break
+
+                if isinstance(page_data, list):
+                    page_data = [
+                        element.decode('utf-8').replace('\x00', ' ') if isinstance(element, bytes) else element
+                        for element in page_data
+                    ]
+                    logger.info(f"NUL characters were found in PR Reviews and replaced with spaces.") 
+                elif isinstance(page_data, bytes):
+                    page_data = page_data.decode('utf-8').replace('\x00', ' ')
+                    logger.info(f"NUL characters were found in PR Reviews and replaced with spaces.") 
+
 
+                pr_reviews.extend(page_data)
 
-            pr_reviews.extend(page_data)
-
-        if pr_reviews:
-            all_pr_reviews[pull_request_id] = pr_reviews
+            if pr_reviews:
+                all_pr_reviews[pull_request_id] = pr_reviews
 
-    if not list(all_pr_reviews.keys()):
-        logger.info(f"{owner}/{repo} No pr reviews for repo")
-        return
+        if not list(all_pr_reviews.keys()):
+            logger.info(f"{owner}/{repo} No pr reviews for repo")
+            return
 
-    contributors = []
-    for pull_request_id in all_pr_reviews.keys():
+        contributors = []
+        for pull_request_id in all_pr_reviews.keys():
 
-        reviews = all_pr_reviews[pull_request_id]
-        for review in reviews:
-            contributor = process_pull_request_review_contributor(review, tool_source, tool_version, data_source)
-            if contributor:
-                contributors.append(contributor)
+            reviews = all_pr_reviews[pull_request_id]
+            for review in reviews:
+                contributor = process_pull_request_review_contributor(review, tool_source, tool_version, data_source)
+                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 = []
-    for pull_request_id in all_pr_reviews.keys():
+        pr_reviews = []
+        for pull_request_id in all_pr_reviews.keys():
 
-        reviews = all_pr_reviews[pull_request_id]
-        for review in reviews:
-
-            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))
+            reviews = all_pr_reviews[pull_request_id]
+            for review in reviews:
+                
+                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)