blob: 27dafb6b3990590998084cacc705fa5b056a72eb (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
import requests
import sys
import time
from typing import Optional
from logger import LOG
class GraphQLClient:
instance = None
url: str = 'https://api.github.com/graphql'
headers: dict
personal_access_token: Optional[str] = None
@classmethod
def setup(cls, pat_token: Optional[str] = None):
if cls.instance is None:
cls.instance = GraphQLClient()
if pat_token is not None:
cls.personal_access_token = pat_token
cls.headers = {
'Authorization': f'Bearer {cls.personal_access_token}',
'Accept': 'application/vnd.github.v3+json',
'Accept': 'application/vnd.github.bane-preview+json',
}
else:
LOG.error('No personal access token provided. Please provide one with the --personal-access-token flag.')
sys.exit(1)
@classmethod
def get_instance(cls):
if cls.instance is None:
raise Exception("The singleton instance has not been set up. Call setup() first.")
return cls.instance
def __init__(self):
if self.instance is not None:
raise Exception("This class is a singleton!")
def make_request(self, query_or_mutation: str, variables: dict):
payload = {
'query': query_or_mutation,
'variables': variables
}
while True:
response = requests.post(self.url, headers=self.headers, json=payload)
data = response.json()
if 'errors' in data:
error_message = data['errors'][0]['message']
if 'was submitted too quickly' in error_message or 'API rate limit exceeded' in error_message or 'Something went wrong while executing your query' in error_message:
LOG.warning('Hit rate limit. Sleeping for 30 seconds...')
time.sleep(30)
continue
else:
LOG.error(f"Fail. Error: {error_message}")
return None
if data.get('extensions', ''):
warning_message = data['extensions']['warnings'][0]['message']
LOG.warning(warning_message)
if 'Bad credentials' in data.get('message', ''):
LOG.error(data['message'])
LOG.error('Invalid personal access token. Please provide a valid one with the --personal-access-token flag.')
sys.exit(1)
return data
|
