"""Copyright since 2019, EPAM Systems This file is part of OSCI. OSCI is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. OSCI is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with OSCI. If not, see .""" from typing import NamedTuple, Optional, List, Any, Dict import datetime import requests import logging import time from requests.structures import CaseInsensitiveDict log = logging.getLogger(__name__) class GithubArchiveRest(requests.Session): BASE_URL = 'https://data.gharchive.org/' @staticmethod def _get_hour_file_name(date: datetime.datetime) -> str: return f'{date.strftime("%Y-%m-%d")}-{date.hour}.json.gz' def _get_hour_url(self, file_name: str) -> str: return self.BASE_URL + file_name def get_hourly_events(self, date: datetime.datetime) -> bytes: log.info(f'Load events for date: {date}') file_name = self._get_hour_file_name(date=date) response = self.get(self._get_hour_url(file_name=file_name)) if response.status_code == 200: return response.content class RequestsLimit(NamedTuple): requests_limit: Optional[int] requests_remaining: Optional[int] limit_reset_time: Optional[datetime.datetime] class GithubRest(requests.Session): """Github Rest Implementation""" GET = 'GET' POST = 'POST' PUT = 'PUT' DELETE = 'DELETE' base_url = 'https://api.github.com' repos_url = base_url + '/repos' def __init__(self, token: str, wait_til_limits: bool = True): """Github api rest session constructor :param token: github api access token :param wait_til_limits: wait reset limits time and retry request """ super().__init__() self.token = token self.limits = RequestsLimit(None, None, None) self.wait_til_limits = wait_til_limits def request(self, method, url, **kwargs) -> requests.Response: """Override request method to patch requests headers and parse response headers""" resp = self.__make_request(method=method, url=url, **kwargs) if self.wait_til_limits and resp.status_code == 403 and self.limits.limit_reset_time is not None: wait: datetime.timedelta = self.limits.limit_reset_time - datetime.datetime.now() log.warning(f'{method} response [{resp.status_code}]' f'remaining_retries={self.limits.requests_remaining} ' f'Wait til {self.limits.limit_reset_time} ({wait})' f'url=`{url}`') time.sleep(wait.total_seconds()) log.debug(f"Retry making request to Github API method={method}, url={url}, kwargs={kwargs} " f"after reset limits") resp = self.__make_request(method=method, url=url, **kwargs) return resp def __make_request(self, method, url, **kwargs) -> requests.Response: log.debug(f"Make request to Github API method={method}, url={url}, kwargs={kwargs}") kwargs['headers'] = {**kwargs.get('headers', {}), **{'Authorization': f'token {self.token}'}} resp = super().request(method=method, url=url, **kwargs) self.limits = self.__get_limits(resp.headers) log.debug(f"Get response[{resp.status_code}] from Github API method={method}, url={url}, kwargs={kwargs}") return resp def _get_repo_url(self, repo_name): """Generates repository url""" return f'{self.repos_url}/{repo_name}' def _get_repo_events_url(self, repo_name): """Generates repository events url""" return f'{self._get_repo_url(repo_name)}/events' @staticmethod def __get_limits(headers) -> RequestsLimit: """Parse limits from response header""" def __parse_int(value: str) -> Optional[int]: """Parse int if not None""" try: return int(value) except Exception as ex: log.warning(f'Error parse `{value}` to int. Exception: {ex}') return def __parse_datetime(value: str) -> Optional[datetime.datetime]: """Parse datetime if not None""" value = __parse_int(value) if value is None: return try: return datetime.datetime.fromtimestamp(value) except Exception as ex: log.warning(f'Error parse `{value}` to datetime. Exception: {ex}') return return RequestsLimit(requests_limit=__parse_int(headers.get('X-RateLimit-Limit')), requests_remaining=__parse_int(headers.get('X-RateLimit-Remaining')), limit_reset_time=__parse_datetime(headers.get('X-RateLimit-Reset'))) def get_repository_events(self, repo_name: str) -> List[dict]: """Get events in repository from API :param repo_name: repository name :return: """ log.info(f'Get events in repository {repo_name}') resp = self.request(method=self.GET, url=self._get_repo_events_url(repo_name=repo_name)) try: if resp.status_code == 200: return resp.json() except Exception as ex: log.warning(f'Exception on parse events in repository {repo_name}: {ex}') log.warning(f'Empty response on events in repository {repo_name}') return [] def get_repository(self, repo_name: str) -> Optional[Dict[str, Any]]: """Get repository information from API :param repo_name: repository name :return: """ log.info(f'Get repository {repo_name} information') resp = self.request(method=self.GET, url=self._get_repo_url(repo_name=repo_name)) try: if resp.status_code == 200: return resp.json() except Exception as ex: log.warning(f'Exception on parse response on getting repository {repo_name}: {ex}') log.warning(f'Empty response on getting repository {repo_name}')