Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def local_schema():
return Client(schema=StarWarsSchema)
def ds():
client = Client(schema=StarWarsSchema)
ds = DSLSchema(client)
return ds
def create_gql_client(token):
transport = RequestsHTTPTransport(url='https://api.github.com/graphql', use_json=True)
transport.headers = { "Authorization": "Bearer {}".format(token) }
client = Client(transport=transport)
return client
'entity': None,
'project': None,
'run': "latest",
'base_url': env.get_base_url("https://api.wandb.ai")
}
if self.api_key is None:
wandb.login()
self.settings.update(overrides)
if 'username' in overrides and 'entity' not in overrides:
wandb.termwarn('Passing "username" to Api is deprecated. please use "entity" instead.')
self.settings['entity'] = overrides['username']
self._projects = {}
self._runs = {}
self._sweeps = {}
self._reports = {}
self._base_client = Client(
transport=RequestsHTTPTransport(
headers={'User-Agent': self.user_agent, 'Use-Admin-Privileges': "true"},
use_json=True,
# this timeout won't apply when the DNS lookup fails. in that case, it will be 60s
# https://bugs.python.org/issue22889
timeout=self._HTTP_TIMEOUT,
auth=("api", self.api_key),
url='%s/graphql' % self.settings['base_url']
)
)
self._client = RetryingClient(self._base_client)
def signal_loading(module, status=False):
if GUI_ENABLED != "true":
return
try:
transport = RequestsHTTPTransport(
url=GRAPHQL_URI,
use_json=True,
headers={
"Content-type": "application/json; charset=utf-8",
"x-hasura-admin-secret": HASURA_GRAPHQL_ACCESS_KEY,
},
verify=False,
)
client = Client(
retries=3, transport=transport, fetch_schema_from_transport=True
)
query = gql(PROCESS_STATES_LOADING_MUTATION)
params = {"name": "{}%".format(module), "loading": status}
client.execute(query, variable_values=params)
except Exception:
log.exception("exception")
def _get_client() -> Client:
if not (GITHUB_API_TOKEN and GITHUB_GRAPHQL_ENDPOINT):
raise GithubGraphqlApiNotProperlyConfiguredException("Token and endpoint properties are required")
# TODO I know it will create the same stuff all the time. Working in progress :)
headers = {"Authorization": f"bearer {GITHUB_API_TOKEN}"}
_transport = RequestsHTTPTransport(url=GITHUB_GRAPHQL_ENDPOINT, use_json=True, timeout=(5, 25), headers=headers)
return Client(retries=3, transport=_transport, fetch_schema_from_transport=True)
def signal_loading(module, status=False):
if GUI_ENABLED != "true":
return
try:
transport = RequestsHTTPTransport(
url=GRAPHQL_URI,
use_json=True,
headers={
"Content-type": "application/json; charset=utf-8",
"x-hasura-admin-secret": HASURA_GRAPHQL_ACCESS_KEY,
},
verify=False,
)
client = Client(
retries=3, transport=transport, fetch_schema_from_transport=True
)
query = gql(PROCESS_STATES_LOADING_MUTATION)
params = {"name": "{}%".format(module), "loading": status}
client.execute(query, variable_values=params)
except Exception:
log.exception("exception")
else:
authorize_success = True
print('OK.')
headers_cachefp = open('headers_cache.json', 'w')
cookies_cachefp = open('cookies_cache.json', 'w')
headers_cachefp.write(ujson.dumps([dict(sess.headers), token_expires]))
cookies_cachefp.write(ujson.dumps(requests.utils.dict_from_cookiejar(request.cookies)))
headers_cachefp.close()
cookies_cachefp.close()
cookies = requests.utils.dict_from_cookiejar(sess.cookies)
headers = sess.headers
return Client(retries=3, transport=CustomTransport(url=endpoint_url, use_json=True, timeout=5,
headers=headers, cookies=cookies),
fetch_schema_from_transport=True)
"cookies": self.cookies,
"timeout": timeout or self.default_timeout,
"verify": self.verify_ssl,
data_key: payload,
}
request = post(self.url, **post_args)
request.raise_for_status()
result = request.json()
assert (
"errors" in result or "data" in result
), 'Received non-compatible response "{}"'.format(result)
return ExecutionResult(errors=result.get("errors"), data=result.get("data"))
class Client(gql.Client):
def __init__(self, url, username, password, **kwargs):
auth = HTTPBasicAuth(username, password)
transport = Transport(url, auth=auth, use_json=True, verify_ssl=False)
super().__init__(
transport=transport, fetch_schema_from_transport=True, **kwargs
)
def paginate(client, query, step):
values = {"first": step}
aggr = rsp = client.execute(query, variable_values=values)
while rsp["locations"]["pageInfo"]["hasNextPage"]:
values["after"] = rsp["locations"]["pageInfo"]["endCursor"]
rsp = client.execute(query, variable_values=values)
aggr["locations"]["edges"].extend(rsp["locations"]["edges"])
return aggr