top_github_repos.py

#!/usr/bin/env python3
import requests
import json
import os
import sys
import datetime
from tenacity import retry, stop_after_attempt, retry_if_exception_type
from string import Template

REPOSITORIES_FOR_LANGUAGE = Template('''
{
  search(query: "language:$language", type: REPOSITORY, first: 100 $after){
    nodes {
      ... on Repository {
        id
      }
    }
    pageInfo {
      hasNextPage,
      endCursor
    }
  }
}
''')

STARGAZERS = Template('''
{
    nodes(ids: ["$nodeid"]) {
        ... on Repository {
            stargazers(first: 100, orderBy: {field: STARRED_AT, direction:DESC} $after) {
                edges {
                    starredAt
                }
                pageInfo {
                    hasNextPage
                    endCursor
                }
            }
        }
	}
}
''')

def generate_graphql_query(language: str='', cursor: str='') -> str:
    if not language:
        raise ValueError('You must specify a language.')
    
    after = cursor and ', after: "{}"'.format(cursor) or ''
    return REPOSITORIES_FOR_LANGUAGE.substitute(
        language=language,
        after=after
    )


def generate_stargazers_graphql_query(nodeid: str='', cursor: str='') -> str:
    if not nodeid:
        raise ValueError('You must specify a nodeid.')
    
    after = cursor and ', after: "{}"'.format(cursor) or ''
    return STARGAZERS.substitute(
        nodeid=nodeid,
        after=after
    )


@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
)
def execute_graphql_query(
    query: str='',
    api_endpoint: str='https://api.github.com/graphql',
    auth_token: str='',
) -> str:
    '''Executes the provided query. Returns a cusor to the next page.'''
    
    if not query:
        raise ValueError('You must specify a query.')
    if not auth_token:
        raise ValueError('You must specify an authorization token.')
    
    headers = {'Authorization': 'token {}'.format(auth_token)}
    data = {'query': query}
    response = requests.post(api_endpoint, headers=headers, json=data)
    response.raise_for_status()
    return response.json()


def extract_node_ids_from_result(result: dict) -> list:
    return [node['id'] for node in result['data']['search']['nodes']]


def extract_cursor_from_result(result: dict) -> str:
    if result['data']['search']['pageInfo']['hasNextPage']:
        return result['data']['search']['pageInfo']['endCursor']
    else:
        return ''


def get_top_repos_for_language(language: str='', limit: int=1000) -> list:
    '''Return a list of GitHub node IDs for repos that highly rank for the
       provided language, up to the provided limit.
    '''
    node_ids = []
    cursor = ''
    while True:
        query = generate_graphql_query(
            language=language,
            cursor=cursor
        )
        result = execute_graphql_query(
            query=query,
            auth_token=os.environ['GITHUB_TOKEN'],
        )
        node_ids += extract_node_ids_from_result(result)
        cursor = extract_cursor_from_result(result)
        if not cursor:
            break
        if len(node_ids) >= limit:
            break

    return node_ids[:limit]

def get_starred_dates_for_repo(nodeid: str='', limit: int=1000) -> list:
    '''Returns a list of date objects corresponding to when the given nodeid
       representing a repository was starred.
    '''
    starred_dates = []
    cursor = ''
    while True:
        query = generate_stargazers_graphql_query(
            nodeid=nodeid,
            cursor=cursor,
        )
        result = execute_graphql_query(
            query=query,
            auth_token=os.environ['GITHUB_TOKEN'],
        )
        starred_dates += extract_starred_at_from_result(result)
        cursor = extract_cursor_from_stargazers_result(result)
        if not cursor:
            break
        if len(starred_dates) >= limit:
            break

    return starred_dates


def extract_starred_at_from_result(result: dict) -> list:
    return [
        datetime.datetime.strptime(node['starredAt'], "%Y-%m-%dT%H:%M:%SZ")
        for node in result['data']['nodes'][0]['stargazers']['edges']
    ]


def extract_cursor_from_stargazers_result(result: dict) -> str:
    if result['data']['nodes'][0]['stargazers']['pageInfo']['hasNextPage']:
        return result['data']['nodes'][0]['stargazers']['pageInfo']['endCursor']
    else:
        return ''


def main():
    pass

if __name__ == '__main__':
    sys.exit(main())