-
-
Save gbaman/b3137e18c739e0cf98539bf4ec4366ad to your computer and use it in GitHub Desktop.
# An example to get the remaining rate limit using the Github GraphQL API. | |
import requests | |
headers = {"Authorization": "Bearer YOUR API KEY"} | |
def run_query(query): # A simple function to use requests.post to make the API call. Note the json= section. | |
request = requests.post('https://api.github.com/graphql', json={'query': query}, headers=headers) | |
if request.status_code == 200: | |
return request.json() | |
else: | |
raise Exception("Query failed to run by returning code of {}. {}".format(request.status_code, query)) | |
# The GraphQL query (with a few aditional bits included) itself defined as a multi-line string. | |
query = """ | |
{ | |
viewer { | |
login | |
} | |
rateLimit { | |
limit | |
cost | |
remaining | |
resetAt | |
} | |
} | |
""" | |
result = run_query(query) # Execute the query | |
remaining_rate_limit = result["data"]["rateLimit"]["remaining"] # Drill down the dictionary | |
print("Remaining rate limit - {}".format(remaining_rate_limit)) |
To pass variables---here queryString and maxItems---you can use format, e.g.
query = """ {{ search(query: "{queryString}", type: REPOSITORY, first: {maxItems}) {{ repositoryCount edges {{ node {{ ... on Repository {{ name url stargazers {{ totalCount }} }} }} }} }} }} """ variables = { 'queryString' : 'is:public stars:>9999', 'maxItems' : '5' } result = run_query(query.format(**variables))
This unfortunately gives me a "KeyError: '\n trip(\n from'"
when parsing this data: https://hastebin.com/buxivileha.makefile
Any idea why?
Because, you don't need to format the string .. you can pass both query and variables as separate keys in the json and the graphql endpoint can figure out what goes were
For anyone trying to pass variables for these queries, I suggest looking at string.Template in python:
from string Import Template
queryTemplate = Template(
"""{
viewer {
repositories(first: $num) {
pageInfo {
hasNextPage
endCursor
}
edges {
node {
name
}
}
}
}
}"""
)
query = queryTemplate.substitute(num=n)
To pass in variables to the query you need to:
- add
variables
parameter in yourrun_query
function. - add
variables
key to the json you that is going to be appended to the graphql endpoint - create the variables dictionary with the parameters you want to pass
my python is a bit rusty but this works
import requests
class graphQL:
headers = {"Authorization": "token <token_here>"}
"""docstring for graphQL"""
def __init__(self):
super(graphQL, self).__init__()
def run_query(self, query, variables): # A simple function to use requests.post to make the API call. Note the json= section.
request = requests.post('https://api.github.com/graphql', json={'query': query, 'variables': variables}, headers=self.headers)
if request.status_code == 200:
return request.json()
else:
raise Exception("Query failed to run by returning code of {}. {}".format(request.status_code, query))
def getClosedIssuesActors(self):
listOfNames = []
query = '''
query($owner: String!, $name: String!) {
repository(owner: $owner, name: $name){
issues(states: CLOSED, first:10){
edges{
node{
... on Issue{
timeline(last: 100){
edges{
node{
__typename
... on ClosedEvent{
actor{
login
}
}
}
}
}
}
}
}
}
}
}'''
variables = {
"owner": "tatmush",
"name": "Saturday-THORN-Dev-Rank"
}
result = self.run_query(query, variables) #execute query
print(result)
don't forget to declare the variables in the query itself.
Thankkk youu!!!!!!!!!!!!!! You saved me:)
Here is an example using generators to overcome GitHub's a limit of 100 items/request.
from requests import exceptions, request
class GitHubQuery:
BASE_URL = "https://api.github.com/graphql"
ORGANIZATION = "org_name"
def __init__(
self,
github_token=None,
query=None,
query_params=None,
additional_headers=None
):
self.github_token = github_token
self.query = query
self.query_params = query_params
self.additional_headers = additional_headers or dict()
@property
def headers(self):
default_headers = dict(
Authorization=f"token {self.github_token}",
)
return {
**default_headers,
**self.additional_headers
}
def generator(self):
while(True):
try:
yield request(
'post',
GitHubQuery.BASE_URL,
headers=self.headers,
json=dict(query=self.query.format_map(self.query_params))
).json()
except exceptions.HTTPError as http_err:
raise http_err
except Exception as err:
raise err
def iterator(self):
pass
import GitHubQuery
class VulnurabilityReport(GitHubQuery):
VULNERABILITY_QUERY = """
{{
organization(login: "org_name") {{
repositories(first: 100, after: {after}) {{
pageInfo{{
hasNextPage
endCursor
}}
totalCount
nodes {{
... on Repository {{
id
name
vulnerabilityAlerts(first: 2) {{
totalCount
nodes {{
id
}}
}}
}}
}}
}}
}}
}}
"""
QUERY_PARAMS = dict(after='')
ADDITIONAL_HEADERS = dict(
Accept="application/vnd.github.vixen-preview+json",
)
def __init__(self, github_token):
super().__init__(
github_token=github_token,
query=VulnurabilityReport.VULNERABILITY_QUERY,
query_params=VulnurabilityReport.QUERY_PARAMS,
additional_headers=VulnurabilityReport.ADDITIONAL_HEADERS
)
def iterator(self):
generator = self.generator()
hasNextPage = True
repos_vulnerabilities = []
while(hasNextPage):
response = next(generator)
endCursor = response["data"]["organization"]["repositories"]["pageInfo"]["endCursor"]
self.query_params = dict(after=endCursor)
repos_vulnerabilities.extend(response["data"]["organization"]["repositories"]["nodes"])
hasNextPage = response["data"]["organization"]["repositories"]["pageInfo"]["hasNextPage"]
return (repos_vulnerabilities)
I just noticed @tatmush use of variables. I used python's format_map, but variables keep the query string simpler.
Hi, same in Python2 (here to check which users as enabled Single Sign-On on an Organization)
Credits for @tatmush and @BarakBD-Globality
class GitHubGraphQLQuery(object):
BASE_URL = "https://api.github.com/graphql"
def __init__(self, token, query, variables=None, additional_headers=None):
self._token = token
self._query = query
self._variables = variables or dict()
self._additional_headers = additional_headers or dict()
@property
def headers(self):
default_headers = dict(Authorization="token {}".format(self._token))
return dict(default_headers.items() + self._additional_headers.items())
def generator(self):
while True:
try:
yield requests.request(
"post",
GitHubGraphQLQuery.BASE_URL,
headers=self.headers,
json={"query": self._query, "variables": self._variables},
).json()
except requests.exceptions.HTTPError as http_err:
raise http_err
except Exception as err:
raise err
def iterator(self):
pass
class GithubSAMLIdentityProvider(GitHubGraphQLQuery):
QUERY = """
query($org: String!, $after: String) {
organization(login: $org) {
samlIdentityProvider {
ssoUrl,
externalIdentities(first: 100, after: $after) {
pageInfo {
hasNextPage
endCursor
}
edges {
node {
guid,
samlIdentity {
nameId
}
user {
login
}
}
}
}
}
}
}
"""
ADDITIONAL_HEADERS = dict(Accept="application/vnd.github.vixen-preview+json")
def __init__(self, organization_name, token):
super(GithubSAMLIdentityProvider, self).__init__(
token=token,
query=GithubSAMLIdentityProvider.QUERY,
variables=dict(org=organization_name, after=None),
additional_headers=GithubSAMLIdentityProvider.ADDITIONAL_HEADERS,
)
self._identities = list()
def iterator(self):
generator = self.generator()
hasNextPage = True
saml_identities = list()
while hasNextPage:
response = next(generator)
endCursor = response["data"]["organization"]["samlIdentityProvider"]["externalIdentities"]["pageInfo"]["endCursor"]
self._variables["after"] = endCursor
saml_identities.extend(
response["data"]["organization"]["samlIdentityProvider"]["externalIdentities"]["edges"]
)
hasNextPage = response["data"]["organization"]["samlIdentityProvider"]["externalIdentities"]["pageInfo"]["hasNextPage"]
return saml_identities
Does anyone if it's possible to define variables inside fragments (see example below)?
fragment repoQuery on Repository($states: [String!]) {
pullRequests(first: 100, states: $states) {
totalCount
nodes{
title
}
}
}
Saved me hours of frustration.... Thank you!
Thanks, exactly what I needed! 👍
👍
Thank you!
Awesome, thanks!
If you're using a Personal Access Token (PAT), you need to put the word "token" or "bearer" before the actual token:
headers = {"Authorization": "token XYZ123TOKENTOKENTOKEN"}
Thanks!
Hey, thanks for this post. Can we enable auto completion or graphiql like features while making these kinds of requests from notebook?
Mutations are the same.
May I mention that the part where it says json={'query': query}
does not vary regardless of whether it is a query or mutation. So, yes, it does end up with {"query": { "query"...
or {"query": {"mutation":...
👍 amazing thank you...just what I was looking for.
This is just what I needed - thanks!
so cool, exactly what I need, thank you!
Easy to read, flexible and still works in 2021, thanks for this!!!!
thanks!
How do I paginate thru results in python code?
How do I paginate thru results in python code?
Hello!!
Hopefully my solution works for you
I did something like this:
def run_query(): # A simple function to use requests.post to make the API call. Note the json= section.
end_cursor = ""
users_list = []
while True:
has_next_page = False
query = """
{
organization(login:"[YOUR_ORG]") {
samlIdentityProvider{
externalIdentities(first:100""" + end_cursor + """){
pageInfo {
startCursor
hasNextPage
endCursor
}
nodes{
samlIdentity{
nameId
username
}
user{
login
id
name
email
}
}
}
}
}
}
"""
request = requests.post('https://api.github.com/graphql', json={'query': query}, headers=headers)
if request.status_code == 200:
result = request.json()
has_next_page = (result["data"]["organization"]
['samlIdentityProvider']
['externalIdentities']['pageInfo']
['hasNextPage'])
new_cursor = (result["data"]["organization"]
['samlIdentityProvider']
['externalIdentities']['pageInfo']
['endCursor'])
result_data = (result["data"]["organization"]
['samlIdentityProvider']
['externalIdentities']['nodes'])
if has_next_page:
end_cursor = ', after:"' + new_cursor + '"'
else:
break
else:
raise Exception("Query failed to run by returning code of {}. {}".format(request.status_code, query))
return
To pass in variables to the query you need to:
- add
variables
parameter in yourrun_query
function.- add
variables
key to the json you that is going to be appended to the graphql endpoint- create the variables dictionary with the parameters you want to pass
my python is a bit rusty but this works
import requests class graphQL: headers = {"Authorization": "token <token_here>"} """docstring for graphQL""" def __init__(self): super(graphQL, self).__init__() def run_query(self, query, variables): # A simple function to use requests.post to make the API call. Note the json= section. request = requests.post('https://api.github.com/graphql', json={'query': query, 'variables': variables}, headers=self.headers) if request.status_code == 200: return request.json() else: raise Exception("Query failed to run by returning code of {}. {}".format(request.status_code, query)) def getClosedIssuesActors(self): listOfNames = [] query = ''' query($owner: String!, $name: String!) { repository(owner: $owner, name: $name){ issues(states: CLOSED, first:10){ edges{ node{ ... on Issue{ timeline(last: 100){ edges{ node{ __typename ... on ClosedEvent{ actor{ login } } } } } } } } } } }''' variables = { "owner": "tatmush", "name": "Saturday-THORN-Dev-Rank" } result = self.run_query(query, variables) #execute query print(result)
don't forget to declare the variables in the query itself.
This approach saved a lot of time for me. thank you for sharing this
query = """ { rateLimit { limit cost remaining resetAt } } """
is not necessary. the ratelimit status is sent in every response.headers
import requests
from datetime import datetime, timezone
import time
github_token = 'ghp_xxxxxxxxxxxxxxxxxxxx'
real_requests_post = requests.post
def wrap_requests_post(*args, **kwargs):
if not 'headers' in kwargs:
kwargs['headers'] = {}
kwargs['headers']['Authorization'] = 'token ' + github_token
response = real_requests_post(*args, **kwargs)
if 'x-ratelimit-used' in response.headers._store:
print("ratelimit status: used %s of %s. next reset in %s minutes" % (
response.headers['X-RateLimit-Used'],
response.headers['X-RateLimit-Limit'],
datetime.fromtimestamp(int(response.headers["X-RateLimit-Reset"]) - time.time(), tz=timezone.utc).strftime("%M:%S")
))
return response
requests.post = wrap_requests_post
query = """
{ viewer { starredRepositories(first: 5) {
nodes { name nameWithOwner sshUrl description }
pageInfo { endCursor hasNextPage }
} } }
"""
response = requests.post('https://api.github.com/graphql', json={'query': query})
data = response.json()
sample output
ratelimit status: used 8 of 5000. next reset in 43:57 minutes
For anyone trying to pass variables for these queries, I suggest looking at string.Template in python:
from string Import Template queryTemplate = Template( """{ viewer { repositories(first: $num) { pageInfo { hasNextPage endCursor } edges { node { name } } } } }""" ) query = queryTemplate.substitute(num=n)
This worked well. thx!
Is there a way to search for all issues across an org? Using REST I would do https://api.github.com/search/issues?q=user:{org_name}&per_page=100&page=1
Any help reformatting this for Graphql would be appreciated.
@els-pnw I was trying to find a way of doing it with graphql explorer but got stuck for this error
"message": "Although you appear to have the correct authorization credentials, the
kubernetes
organization has enabled OAuth App access restrictions, meaning that data access to third-parties is limited. For more information on these restrictions, including how to enable this app, visit https://docs.github.com/articles/restricting-access-to-your-organization-s-data/"
I ended up with something like this
query ($org_name: String!) {
organization(login: $org_name) {
repositories(last: 3) {
nodes {
issues(last: 2) {
nodes {
number
assignees {
edges {
node {
name
}
}
}
author {
login
}
state
}
}
}
}
}
}
Not sure how much it will help you, again the explorer is nice, just play with it, hope you will get your answer.
To pass variables---here queryString and maxItems---you can use format, e.g.