How to use the gql.Client function in gql

To help you get started, we’ve selected a few gql examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github graphql-python / gql / tests / starwars / test_validation.py View on Github external
def local_schema():
    return Client(schema=StarWarsSchema)
github graphql-python / gql / tests / starwars / test_dsl.py View on Github external
def ds():
    client = Client(schema=StarWarsSchema)
    ds = DSLSchema(client)
    return ds
github mavlink / MAVSDK / tools / generate_changelog.py View on Github external
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
github wandb / client / wandb / apis / public.py View on Github external
'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)
github FORTH-ICS-INSPIRE / artemis / monitor / core / utils / __init__.py View on Github external
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")
github willianantunes / flask-playground / flask_playground / services / github_graphql_api.py View on Github external
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)
github FORTH-ICS-INSPIRE / artemis / backend / core / utils / __init__.py View on Github external
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")
github itolosa / orionx-api-client / orionxapi / future / connection_manager.py View on Github external
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)
github magma / magma / symphony / tools / everis / pager / pager.py View on Github external
"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