github-actions-bigquery.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Github Actions\n",
    "\n",
    "Acts as a client to the Github API, and pulls information gathering user actions over time. Data can then be upload to a database for analysis. This extraction is tightly coupled to an Apache Superset dashboard the allows for visualizing key repository metrics.\n",
    "\n",
    "For reference, see this related nice [Github entity relationship diagram (ERD)](https://docs.google.com/presentation/d/e/2PACX-1vRDd0-PUMa7Jg4O5mNcaDwW3FndDOQZG7ZximXWRgxpyeCnUHSJl2-ZF9HWfktvghLUV8WYp7B8VNXL/embed?start=false&amp;loop=false&amp;delayms=3000&slide=id.g244d368397_0_1) provided on Fivetran's website.\n",
    "\n",
    "\n",
    "### TODO\n",
    "* Includes logic to do differential updates"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import Image\n",
    "Image(filename='dash.png') "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Thanks to Fivetran for this great ERD of the Github API\n",
    "Image(filename='github_erd.png') "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime, timedelta\n",
    "import logging\n",
    "import os\n",
    "import pickle\n",
    "import time\n",
    "import json\n",
    "from dateutil import parser\n",
    "\n",
    "from github import Github\n",
    "from github.Label import Label\n",
    "from github.GithubException import RateLimitExceededException\n",
    "\n",
    "import pandas as pd\n",
    "import sqlalchemy as sqla\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set / reset the \n",
    "cache = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "REPO = \"apache/incubator-superset\"\n",
    "\n",
    "GITHUB_TOKENS = json.loads(os.environ['GITHUB_TOKENS'])\n",
    "#GITHUB_TOKENS = None  # str or list[str], also will try to use an env var GITHUB_TOKEN\n",
    "RATE_LIMIT_SLEEP = 15\n",
    "ITEMS_LIMIT = 100000\n",
    "LOG_AT_EVERY = 50\n",
    "RATE_LIMIT_BUFFER = 500  # pause when less than that is left\n",
    "RATE_LIMIT_CHECK_INTERVAL = 50\n",
    "DB_URI = 'mysql://root@localhost/examples'\n",
    "BOT_LIST = {\n",
    "    'codecov-io',\n",
    "}\n",
    "def IS_BOT(username):\n",
    "    if username and '[bot]' in username:\n",
    "        return True\n",
    "    \n",
    "    return username in BOT_LIST\n",
    "\n",
    "# Cutom logic related to Superset org-assignement \n",
    "ORG_MAP = {\n",
    "    'Airbnb': [\n",
    "        'john-bodley',\n",
    "        'graceguo-supercat',\n",
    "        'kristw', 'vera-liu',\n",
    "        'michellethomas',\n",
    "        'timifasubaa',\n",
    "        'williaster',\n",
    "        'etr2460',\n",
    "    ],\n",
    "    'Preset': [\n",
    "        'mistercrunch',\n",
    "        'villebro', 'dpgaspar', 'willbarrett',\n",
    "        'rusackas', 'craig-rueda', 'nytai',\n",
    "        'robdiciuccio',\n",
    "    ],\n",
    "    'Lyft': ['betodealmeida', 'hughhhh', 'khtruong'],\n",
    "}\n",
    "# Reversing the lookup\n",
    "ORG_LKP = {}\n",
    "for org, users in ORG_MAP.items():\n",
    "    for u in users:\n",
    "        ORG_LKP[u] = org\n",
    "        \n",
    "def ATTRIBUTE_ORG_LAMBDA(actor, dttm):\n",
    "    \"\"\"Used to define custom organization attribution logic\"\"\"\n",
    "    if actor == 'mistercrunch' and dttm:\n",
    "        if dttm <= datetime(2017, 10, 1):\n",
    "            return 'Airbnb'\n",
    "        elif dttm <= datetime(2018, 12, 1):\n",
    "            return 'Lyft'\n",
    "    return ORG_LKP.get(actor, 'Other')\n",
    "\n",
    "def get_org_repo_names(org):\n",
    "    ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "    gh = ghc._get_github_instance()\n",
    "    org = gh.get_organization(org)\n",
    "    repos = org.get_repos()\n",
    "    return [repo.full_name for repo in repos if not repo.fork]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class GithubActionClient:\n",
    "    version = 0.11  # bumping the version invalidates entries in the cache\n",
    "    def __init__(self, repo, cache=None, api_tokens=None, attribute_org_func=None):\n",
    "        \"\"\"\n",
    "        A client that connects to the Github API and gets all user actions\n",
    "        \n",
    "        At a high level, the client receives a cache object that can be managed externally to\n",
    "        be longer-lived than the client object, and accumulates actions as it scans the\n",
    "        different types of objects (issues, stars, watchers, ...).\n",
    "        \n",
    "        The `process_all` method is probably the only one you'll need to call, look at its\n",
    "        docstring for more details as to how to make the client work for you. It can\n",
    "        be parameterized to run in backfill or incremental mode.\n",
    "        \n",
    "        Note that it's fairly common to hit the rate limit (currently at 5k/hour) and that\n",
    "        the lib supports receiving a list of tokens and will take advantage of that. It's\n",
    "        fairly relentless and usually tends to restart until it makes it through.\n",
    "        \n",
    "        :param repo: a string representing the Github repository (ie: apache/incubator-superset)\n",
    "        :type repo: str\n",
    "        :param cache: (optional) a dict to use as a cache, this enables making the cache\n",
    "          global to the notebook and passed back as the client is re-initialized\n",
    "        :type cache: dict\n",
    "        :param api_tokens: (optional) one or many Github API token key, if not specified, we'll\n",
    "          look and a `GITHUB_TOKEN` envrironment variable \n",
    "        :type api_tokens: str or list[str]\n",
    "        :param attribute_org_func: a function or lambda that receives the actor and datetime\n",
    "          of the event and returns an org\n",
    "        :type attribute_org_func:\n",
    "        \"\"\"\n",
    "        self.cache = {} if cache is None else cache\n",
    "        self.repo = repo\n",
    "        self.handle_api_tokens(api_tokens)\n",
    "        self._attribute_org_func = attribute_org_func\n",
    "        \n",
    "        # stats\n",
    "        self.skipped = 0\n",
    "        self.errors = 0\n",
    "        self.processed = 0\n",
    "        self.last_updated_at = None\n",
    "        self.last_id_processed = None\n",
    "        \n",
    "    def reset_cache(self):\n",
    "        self.cache = {}\n",
    "        \n",
    "    def handle_api_tokens(self, api_tokens):\n",
    "        self._api_token_index = 0\n",
    "        if api_tokens:\n",
    "            self.api_tokens = api_tokens\n",
    "        else:\n",
    "            # the token currently in use\n",
    "            envvar_content = os.environ.get(\"GITHUB_TOKENS\") or os.environ.get(\"GITHUB_TOKEN\")\n",
    "            self.api_tokens = envvar_content.split(',')\n",
    "        \n",
    "    @property\n",
    "    def _current_token(self):\n",
    "        return self.api_tokens[self._api_token_index]\n",
    "    \n",
    "    def _get_github_instance(self):\n",
    "        return Github(self._current_token, per_page=100)\n",
    "\n",
    "    def get_repo(self):\n",
    "        g = self._get_github_instance()\n",
    "        return g.get_repo(self.repo)\n",
    "    \n",
    "    def dump_cache(self, filename=None):\n",
    "        dt = datetime.now().isoformat().replace('-','_').replace(':', '_')\n",
    "        filename = filename or f'./cachedump_{dt}.pickle'\n",
    "        print(f'Dumping to: {filename}')\n",
    "        with open(filename, 'wb') as f: \n",
    "            pickle.dump(self.cache, f)\n",
    "    \n",
    "    def load_cache(self, filename):\n",
    "        with open(filename, 'rb') as f: \n",
    "            self.cache = pickle.load(f)\n",
    "\n",
    "    def _log_progress(self, i, force=False):\n",
    "        if (i and i % LOG_AT_EVERY == 0) or force:\n",
    "            cache_keys = [s for _, s, _ in self.cache.keys()]\n",
    "            print(\n",
    "                f\"{datetime.now().isoformat()[10:16]} | \"\n",
    "                f\"{i} | \"\n",
    "                f\"PRs:{cache_keys.count('pr')} | \"\n",
    "                f\"issues:{cache_keys.count('issue')} | \"\n",
    "                f\"stars:{cache_keys.count('star')} | \"\n",
    "                f\"skipped: {self.skipped} | \"\n",
    "                f\"processed: {self.processed} | \"\n",
    "                f\"errors: {self.errors} | \"\n",
    "                #f\"last_id: {self.last_id_processed} | \"\n",
    "                #f\"last_updated: {self.last_updated_at} | \"\n",
    "            )\n",
    "\n",
    "    def _process_collection(self, key_lambda, get_colection_func, object_processor, overwrite_cache=False):\n",
    "        \"\"\"A generic function to process a collection of things\n",
    "        \n",
    "        It works in a similar manner whether the collection is stars, issues, watches or else, handles\n",
    "        logging, errors and rate limit issues consistantly.\"\"\"\n",
    "        l = []\n",
    "        last_i = None\n",
    "        for i, obj in enumerate(get_colection_func()):\n",
    "            l.append(obj)\n",
    "            if i > ITEMS_LIMIT:\n",
    "                break\n",
    "            self._log_progress(i)\n",
    "            if i % RATE_LIMIT_CHECK_INTERVAL == 0:\n",
    "                self.check_rate_limit()\n",
    "            k = key_lambda(obj)\n",
    "            \n",
    "            if (\n",
    "                k not in self.cache or\n",
    "                (\n",
    "                    self.cache[k][0].get('version') != self.version\n",
    "                    or (\n",
    "                        overwrite_cache \n",
    "                        and datetime.now() - self.cache[k][0].get('processed_on', datetime(1979, 7, 19)) > timedelta(days=1)\n",
    "                    )\n",
    "                )\n",
    "            ):\n",
    "                try:\n",
    "                    result = object_processor(obj)\n",
    "                    self.cache[k] = result\n",
    "                    self.processed += 1\n",
    "                except Exception as e:\n",
    "                    self.errors +=1\n",
    "                    logging.exception(e)\n",
    "            else:\n",
    "                self.skipped += 1\n",
    "            if hasattr(obj, 'updated_at'):\n",
    "                self.last_updated_at = obj.updated_at\n",
    "            if hasattr(obj, 'id'):\n",
    "                self.last_id_processed = obj.id\n",
    "            last_i = i\n",
    "        # Always log progress when done\n",
    "        self._log_progress(last_i, force=True)\n",
    "        return l\n",
    "\n",
    "    def process_issues(self, overwrite_cache=False, since=None):\n",
    "        repo = self.get_repo()\n",
    "        DIRECTION = 'desc'\n",
    "        if since:\n",
    "            get_issues_func = lambda: repo.get_issues(state='all', since=since, direction=DIRECTION)\n",
    "        else:\n",
    "            get_issues_func = lambda: repo.get_issues(state='all', direction=DIRECTION)\n",
    "        self._process_collection(lambda x: (self.repo, 'pr' if x.pull_request else 'issue', x.number), get_issues_func, self._process_issue, overwrite_cache)\n",
    "\n",
    "    def _process_star(self, star):\n",
    "        login = star.user.login\n",
    "        o = self._create_action('star', star.starred_at, login, login)\n",
    "        return [o]\n",
    "\n",
    "    def process_stars(self, overwrite_cache=False, since=None):\n",
    "        repo = self.get_repo()\n",
    "        return self._process_collection(lambda x: (self.repo, 'star', x.user.login), lambda: repo.get_stargazers_with_dates(), self._process_star, overwrite_cache)\n",
    "\n",
    "    def _process_fork(self, fork):\n",
    "        login = fork.owner.login if fork.owner else None\n",
    "        o = self._create_action('fork', fork.created_at, login, login)\n",
    "        return [o]\n",
    "    \n",
    "    def process_forks(self, overwrite_cache=False, since=None):\n",
    "        repo = self.get_repo()\n",
    "        self._process_collection(lambda x: (self.repo, 'fork', x.owner.login if x.owner else None), lambda: repo.get_forks(), self._process_fork, overwrite_cache)\n",
    "\n",
    "    def _process_watcher(self, watcher):\n",
    "        login = watcher.login\n",
    "        o = self._create_action('watch', watcher.created_at, login, login)\n",
    "        return [o]\n",
    "    \n",
    "    def process_watchers(self, overwrite_cache=False, since=None):\n",
    "        repo = self.get_repo()\n",
    "        self._process_collection(lambda x: (self.repo, 'watch', x.login), lambda: repo.get_watchers(), self._process_watcher, overwrite_cache)\n",
    "\n",
    "    def _process_reaction(self, reaction, labels, title=None):\n",
    "        return self._create_action('reaction_created', reaction.created_at, reaction.user.login, reaction.id, labels, type_=reaction.content, title=title)\n",
    "\n",
    "    def _process_comment(self, comment, labels, title):\n",
    "        actions = []\n",
    "        actions.append(self._create_action('comment_created', comment.created_at, comment.user.login, comment.id, labels, title=title))\n",
    "        for reaction in comment.get_reactions():\n",
    "            actions.append(self._process_reaction(reaction, labels, title))\n",
    "        return actions\n",
    "\n",
    "    def _process_issue(self, issue):\n",
    "        \"\"\"Process an issues, whether it's a PR or an actual issue\"\"\"\n",
    "        actions = []\n",
    "        labels = '|'.join([lbl.name for lbl in issue.get_labels()])\n",
    "        labels = f'|{labels}|'\n",
    "        title = issue.title\n",
    "        for reaction in issue.get_reactions():\n",
    "            actions.append(self._process_reaction(reaction, labels))\n",
    "            \n",
    "        closed_at = issue.closed_at\n",
    "        closed_by = issue.closed_by.login if issue.closed_by else None\n",
    "\n",
    "        if issue.pull_request:\n",
    "            # If the issue is a pull request\n",
    "            pull = issue.as_pull_request()\n",
    "            #action, dttm, actor, id_=None, labels=None, type_=None, duration=None,\n",
    "            pr_action_kwargs = {\n",
    "                \"pr_comments\": pull.comments,\n",
    "                \"pr_review_comments\": pull.review_comments,\n",
    "                \"pr_commits\": pull.commits,\n",
    "                \"pr_additions\": pull.additions,\n",
    "                \"pr_deletions\": pull.deletions,\n",
    "                \"pr_changed_files\": pull.changed_files,\n",
    "                \"closed_at\": closed_at,\n",
    "                \"closed_by\": closed_by,\n",
    "                \"title\": title,\n",
    "                \"id_\": pull.number,\n",
    "                \"labels\": labels,\n",
    "            }\n",
    "            o = self._create_action('pr_created', pull.created_at, pull.user.login, **pr_action_kwargs)\n",
    "            actions.append(o)\n",
    "\n",
    "            if pull.merged_at:\n",
    "                duration = (pull.merged_at - pull.created_at).total_seconds()\n",
    "                actions.append(self._create_action('pr_merged', pull.merged_at, pull.merged_by.login, duration=duration, **pr_action_kwargs))\n",
    "            if pull.closed_at:\n",
    "                duration = (pull.closed_at - pull.created_at).total_seconds()\n",
    "                actions.append(self._create_action('pr_closed', pull.closed_at, pull.user.login, **pr_action_kwargs))\n",
    "\n",
    "            for review in pull.get_reviews():\n",
    "                login = review.user.login if review.user else None\n",
    "                try:\n",
    "                    dttm = review.submitted_at\n",
    "                except Exception as e:\n",
    "                    # TODO research corner case on a single PR / review\n",
    "                    continue\n",
    "                id_ = review.id\n",
    "                actions.append(self._create_action('pr_review', dttm, login, id_, labels, title=title, closed_at=closed_at, closed_by=closed_by))\n",
    "        else:\n",
    "            actions.append(self._create_action('issue_created', issue.created_at, issue.user.login, issue.number, labels, title=title))\n",
    "            if issue.closed_at:\n",
    "                duration = (issue.closed_at - issue.created_at).total_seconds()\n",
    "                actions.append(self._create_action('issue_closed', issue.closed_at, issue.user.login, issue.number, labels, duration=duration, title=title))\n",
    "            \n",
    "        for comment in issue.get_comments():\n",
    "            actions += self._process_comment(comment, labels, title)\n",
    "        for reaction in issue.get_reactions():\n",
    "            actions.append(self._process_reaction(reaction, labels, title=title))\n",
    "        return actions\n",
    "\n",
    "    def _create_action(\n",
    "            self,\n",
    "            action,\n",
    "            dttm,\n",
    "            actor,\n",
    "            id_=None,\n",
    "            labels=None, type_=None, duration=None, title=None, closed_at=None, closed_by=None,\n",
    "            pr_comments=None,\n",
    "            pr_review_comments=None,\n",
    "            pr_commits=None,\n",
    "            pr_additions=None,\n",
    "            pr_deletions=None,\n",
    "            pr_changed_files=None,\n",
    "    ):\n",
    "        return {\n",
    "            \"pr_comments\": pr_comments,\n",
    "            \"pr_review_comments\": pr_review_comments,\n",
    "            \"pr_commits\": pr_commits,\n",
    "            \"pr_additions\": pr_additions,\n",
    "            \"pr_deletions\": pr_deletions,\n",
    "            \"pr_changed_files\": pr_changed_files,\n",
    "            'action': action,\n",
    "            'dttm': dttm,\n",
    "            'actor': actor,\n",
    "            'labels': labels,\n",
    "            'id': id_,\n",
    "            'type': type_,\n",
    "            'duration': duration,\n",
    "            'org': self._attribute_org_func(actor, dttm) if self._attribute_org_func else None,\n",
    "            'title': title,\n",
    "            'closed_at': closed_at,\n",
    "            'closed_by': closed_by,\n",
    "            'is_bot': IS_BOT(actor),\n",
    "            'version': self.version,\n",
    "            'processed_on': datetime.now(),\n",
    "        }\n",
    "\n",
    "    def get_rate_limit(self):\n",
    "        gh = self._get_github_instance()\n",
    "        rl = gh.get_rate_limit()\n",
    "        return rl.core\n",
    "    \n",
    "    def _switch_token(self):\n",
    "        if len(self.api_tokens) > 1:\n",
    "            self._api_token_index += 1\n",
    "            if self._api_token_index >= len(self.api_tokens):\n",
    "                self._api_token_index = 0\n",
    "            print(f\"Switching to using token #{self._api_token_index}\")\n",
    "            return True\n",
    "        return False\n",
    "\n",
    "    def check_rate_limit(self, sleep=True):\n",
    "        rate_limit = self.get_rate_limit()\n",
    "        while rate_limit.remaining < RATE_LIMIT_BUFFER:\n",
    "            print(f\"[rate limit] {rate_limit.remaining} remaining, buffer is {RATE_LIMIT_BUFFER}, reset: {rate_limit.reset}\")\n",
    "            if sleep:\n",
    "                print(f\"Sleeping for {RATE_LIMIT_SLEEP} seconds ...\")\n",
    "                time.sleep(RATE_LIMIT_SLEEP)\n",
    "            self._switch_token()\n",
    "            rate_limit = self.get_rate_limit()\n",
    "\n",
    "    def process_all(self, overwrite_cache=False, since=None, issues=True, stars=True, forks=True, watchers=True):\n",
    "        \"\"\"Process everything!\n",
    "        \n",
    "        Two main operation mode here: incremental (using the `since` parameter, usually in combination with\n",
    "        the overwrite_cache=True) or more of a backfill/first load approach where you typically just\n",
    "        call `process_all()`. On large repository, this is likely to take a long time (hours), and it's\n",
    "        been known to fail half way (closing your laptop, network disconnect, ...). Restarting the process\n",
    "        should leverage the cache and go much faster to get to where it left off.\n",
    "        \"\"\"\n",
    "        # Check if we're good to go\n",
    "        self.check_rate_limit(sleep=True)\n",
    "        done = False\n",
    "        while not done:\n",
    "            try:\n",
    "                if stars:\n",
    "                    print(f\"{self.repo} - Processing stars\")\n",
    "                    self.process_stars(overwrite_cache, since)\n",
    "                if issues:\n",
    "                    print(f\"{self.repo} - Processing issues\")\n",
    "                    self.process_issues(overwrite_cache, since)\n",
    "                if forks:\n",
    "                    print(f\"{self.repo} - Processing forks\")\n",
    "                    self.process_forks(overwrite_cache, since)\n",
    "                if watchers:\n",
    "                    print(\"Processing watchers\")\n",
    "                    print(f\"{self.repo} - Processing watchers\")\n",
    "                    self.process_watchers(overwrite_cache, since)\n",
    "                done = True\n",
    "            except RateLimitExceededException as e:\n",
    "                print(f\"Hit rate limit... remaining: {self.get_rate_limit()}\")\n",
    "                self.check_rate_limit()\n",
    "            except Exception as e:\n",
    "                logging.error(\"Uh oh - unhandled error, restarting process_all...\")\n",
    "            \n",
    "    def get_df_from_cache(self):\n",
    "        \"\"\"Returns a pandas dataframe from the cache\"\"\"\n",
    "        rows = []\n",
    "        for (repo, object_type, id_), action_list in self.cache.items():\n",
    "            for action in action_list:\n",
    "                action.update({\n",
    "                    'repo': repo,\n",
    "                    'parent_type': object_type,\n",
    "                    'parent_id': id_,\n",
    "                })\n",
    "                rows.append(action)\n",
    "        df = pd.DataFrame(rows)\n",
    "        return df\n",
    "    \n",
    "    @property\n",
    "    def target_table_dtypes(self):\n",
    "        d = dict(\n",
    "            index='BIGINT',\n",
    "            action='STRING',\n",
    "            actor='STRING',\n",
    "            labels='STRING',\n",
    "            id='STRING',\n",
    "            type='STRING',\n",
    "            duration='BIGINT',\n",
    "            org='STRING',\n",
    "            title='STRING',\n",
    "            repo='STRING',\n",
    "            parent_type='STRING',\n",
    "            parent_id='STRING',\n",
    "        )\n",
    "        return None\n",
    "        return [dict(name=k, type=v) for k, v in d.items()]\n",
    "    \n",
    "    def _create_engine(self):\n",
    "        return sqla.create_engine(DB_URI, credentials_path='/Users/max/.dbt/dbt-user-creds.json')\n",
    "    \n",
    "    def _target_table(self, incremental=False):\n",
    "        SCHEMA = 'stg_github'\n",
    "        return  f'{SCHEMA}.github_actions'\n",
    "        return f'{SCHEMA}.tmp_github_actions' if incremental else f'{SCHEMA}.github_actions'\n",
    "    \n",
    "    def to_sql(self, incremental=False, **kwargs):\n",
    "        \"\"\"Flushes to database, takes all arguments that Pandas' to_sql takes\"\"\"\n",
    "        df = self.get_df_from_cache()\n",
    "        conn = self._create_engine()\n",
    "        table_name = self._target_table(incremental)\n",
    "        print(f\"Loading into {table_name}\")\n",
    "        df.to_gbq(table_name, \"preset-cloud-analytics\", if_exists='replace', table_schema=self.target_table_dtypes, location='us-west2', **kwargs)\n",
    "        \n",
    "    def to_sql_chunked(self, incremental=False, chunksize=100, start_row=0, **kwargs):\n",
    "        \"\"\"Flushes to database, takes all arguments that Pandas' to_sql takes\"\"\"\n",
    "        df = self.get_df_from_cache()\n",
    "        conn = self._create_engine()\n",
    "        table_name = self._target_table(incremental)\n",
    "        print(f\"Loading into {table_name}\")\n",
    "        \n",
    "        def chunker(seq, size, start_row=0):\n",
    "            return ((seq[pos:pos + size], pos) for pos in range(start_row, len(seq), size))\n",
    "        \n",
    "        for tdf, i in chunker(df, chunksize, start_row):\n",
    "            print(f\"Loading rows ({i}, {i+chunksize-1})\")\n",
    "            if_exists = 'replace' if i == 0 else 'append'\n",
    "            location = 'us-west2' if i == 0 else None\n",
    "            tdf.to_gbq(table_name, \"preset-cloud-analytics\", if_exists=if_exists, table_schema=self.target_table_dtypes, location=location, **kwargs)\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run the process for a list of repos\n",
    "SINCE = datetime(2021, 8, 1)\n",
    "SINCE = None\n",
    "\n",
    "OVERWRITE = True if SINCE else False\n",
    "PROCESS_STARS = False if SINCE else True\n",
    "PROCESS_FORKS = False if SINCE else True\n",
    "PROCESS_STARS = False\n",
    "PROCESS_FORKS = False\n",
    "WATCHERS = False\n",
    "BAR = (\"-\" * 40) + \"\\n\"\n",
    "repos =  [\n",
    "    'apache/incubator-superset',\n",
    "    'apache-superset/superset-ui',\n",
    "    'apache-superset/superset-ui-plugins',\n",
    "] + get_org_repo_names('preset-io')\n",
    "\n",
    "for repo in repos:\n",
    "    print(f\"{BAR}Processing repo: {repo}\\n{BAR}\")\n",
    "    ghc = GithubActionClient(repo, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "    ghc.process_all(since=SINCE, overwrite_cache=OVERWRITE, stars=PROCESS_STARS, watchers=WATCHERS, forks=PROCESS_FORKS)\n",
    "    #ghc.process_stars()\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Push to the database\n",
    "DB_URI = 'bigquery://preset-cloud-analytics'\n",
    "\n",
    "ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "print(len(ghc.get_df_from_cache()))\n",
    "#ghc.to_sql(incremental=False)\n",
    "ghc.to_sql_chunked(incremental=False, chunksize=25000, start_row=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "upsert_sql = \"\"\"\n",
    "SELECT \n",
    "    a.index,\n",
    "    a.action,\n",
    "    a.dttm,\n",
    "    a.actor,\n",
    "    a.labels,\n",
    "    a.id,\n",
    "    a.type,\n",
    "    a.duration,\n",
    "    a.org,\n",
    "    a.title,\n",
    "    a.repo,\n",
    "    a.parent_type,\n",
    "    a.parent_id\n",
    "FROM github_actions a\n",
    "LEFT JOIN tmp_github_actions b ON CAST(a.id AS TEXT) = CAST(b.id AS TEXT) AND a.action = b.action\n",
    "WHERE b.dttm IS NULL\n",
    "UNION ALL\n",
    "SELECT\n",
    "    index,\n",
    "    action,\n",
    "    dttm,\n",
    "    actor,\n",
    "    labels,\n",
    "    id,\n",
    "    type,\n",
    "    duration,\n",
    "    org,\n",
    "    title,\n",
    "    repo,\n",
    "    parent_type,\n",
    "    parent_id\n",
    "FROM tmp_github_actions\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "\n",
    "ghc.load_cache('./cachedump_2021_08_28T22_26_46.204566.pickle')\n",
    "cache = ghc.cache"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "#ghc.dump_cache()\n",
    "\n",
    "'''\n",
    "comments = ,\n",
    "            review_comments\": 0,\n",
    "              \"maintainer_can_modify\": true,\n",
    "              \"commits\": 3,\n",
    "              \"additions\": 100,\n",
    "              \"deletions\": 3,\n",
    "              \"changed_files\": 5'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "repo = ghc.get_repo()\n",
    "pull = repo.get_pull(16410)\n",
    "pull.review_comments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ghc = GithubActionClient(REPO, cache, GITHUB_TOKENS, ATTRIBUTE_ORG_LAMBDA)\n",
    "repo = ghc.get_repo()\n",
    "pulls = repo.get_pulls()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}