68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""GitHub API module"""
|
|
import logging
|
|
import requests
|
|
from datetime import datetime
|
|
from time import sleep
|
|
from requests.adapters import HTTPAdapter
|
|
from requests.packages.urllib3.util.retry import Retry
|
|
|
|
|
|
class GitHubApi:
|
|
def __init__(self, token=None, headers=None, proxy=None):
|
|
self.base_url = 'https://api.github.com'
|
|
self.session = requests.Session()
|
|
|
|
retries = Retry(total=3, backoff_factor=1)
|
|
self.session.mount('https://', HTTPAdapter(max_retries=retries))
|
|
|
|
if token:
|
|
self.session.headers.update({'Authorization': f'Bearer {token}'})
|
|
if headers:
|
|
self.session.headers.update(headers)
|
|
if proxy:
|
|
self.session.proxies.update(proxy)
|
|
|
|
# Rate limits
|
|
self.rate_reset = None
|
|
self.rate_remaining = None
|
|
|
|
def __get(self, endpoint, params=None):
|
|
"""GET an HTTP resource and return JSON."""
|
|
url = self.base_url + endpoint
|
|
|
|
# Check rate limits
|
|
if self.rate_remaining == 0 and self.rate_reset > datetime.now():
|
|
logging.warning(f'Out of API calls. Sleeping until {self.rate_reset}.')
|
|
delta = self.rate_reset - datetime.now()
|
|
sleep(delta.total_seconds())
|
|
|
|
try:
|
|
response = self.session.get(url, params=params)
|
|
except requests.RequestException as e:
|
|
logging.warning(f'Failed to execute GET request: {str(e)}')
|
|
return None
|
|
|
|
# 304 Not Modified
|
|
if response.status_code == '304':
|
|
return None
|
|
|
|
self.__parse_headers(response.headers)
|
|
return response.json()
|
|
|
|
def __parse_headers(self, headers):
|
|
"""Parse headers to update rate limits."""
|
|
self.rate_reset = datetime.fromtimestamp(int(headers['x-ratelimit-reset']))
|
|
self.rate_remaining = int(headers['x-ratelimit-remaining'])
|
|
|
|
def get_events(self, page=None):
|
|
"""Fetch GitHub events."""
|
|
endpoint = '/events'
|
|
params = None
|
|
|
|
if page:
|
|
params = {'page': page}
|
|
|
|
events = self.__get(endpoint, params)
|
|
|
|
return events
|