#default_exp core
Detailed information on the GhApi API
#export
from fastcore.utils import *
from fastcore.foundation import *
from fastcore.meta import *
from ghapi.metadata import funcs
import mimetypes,base64
from inspect import signature,Parameter,Signature
from urllib.request import Request
from urllib.error import HTTPError
from datetime import datetime,timedelta
from pprint import pprint
#hide
from nbdev import *
#export
GH_HOST = "https://api.github.com"
_DOC_URL = 'https://docs.github.com/'
#export
def _preview_hdr(preview): return {'Accept': f'application/vnd.github.{preview}-preview+json'} if preview else {}
def _mk_param(nm, **kwargs): return Parameter(nm, kind=Parameter.POSITIONAL_OR_KEYWORD, **kwargs)
def _mk_sig_detls(o):
res = {}
if o[0]!=object: res['annotation']=o[0]
res['default'] = o[1] if len(o)>1 else None
return res
def _mk_sig(req_args, opt_args, anno_args):
params = [_mk_param(k) for k in req_args]
params += [_mk_param(k, default=v) for k,v in opt_args.items()]
params += [_mk_param(k, **_mk_sig_detls(v)) for k,v in anno_args.items()]
return Signature(params)
#export
class _GhObj: pass
#export
class _GhVerb(_GhObj):
__slots__ = 'path,verb,tag,name,summary,url,route_ps,params,data,preview,client,__doc__'.split(',')
def __init__(self, path, verb, oper, summary, url, params, data, preview, client, kwargs):
tag,name = oper.split('/')
name = name.replace('-','_')
path,_,_ = partial_format(path, **kwargs)
route_ps = stringfmt_names(path)
__doc__ = summary
data = {o[0]:o[1:] for o in data}
store_attr()
def __call__(self, *args, headers=None, **kwargs):
headers = {**_preview_hdr(self.preview),**(headers or {})}
d = list(self.data)
flds = [o for o in self.route_ps+self.params+d if o not in kwargs]
for a,b in zip(args,flds): kwargs[b]=a
kwargs = {k:v for k,v in kwargs.items() if v is not None}
route_p,query_p,data_p = [{p:kwargs[p] for p in o if p in kwargs}
for o in (self.route_ps,self.params,d)]
return self.client(self.path, self.verb, headers=headers, route=route_p, query=query_p, data=data_p)
def __str__(self): return f'{self.tag}.{self.name}{signature(self)}\n{self.doc_url}'
@property
def __signature__(self): return _mk_sig(self.route_ps, dict.fromkeys(self.params), self.data)
__call__.__signature__ = __signature__
@property
def doc_url(self): return _DOC_URL + self.url.replace(" ","_")
def _repr_markdown_(self):
params = ', '.join(self.route_ps+self.params+list(self.data))
return f'[{self.tag}.{self.name}]({self.doc_url})({params}): *{self.summary}*'
__repr__ = _repr_markdown_
#export
class _GhVerbGroup(_GhObj):
def __init__(self, name, verbs):
self.name,self.verbs = name,verbs
for o in verbs: setattr(self, o.name, o)
def __str__(self): return "\n".join(str(v) for v in self.verbs)
def _repr_markdown_(self): return "\n".join(f'- {v._repr_markdown_()}' for v in self.verbs)
#export
_docroot = 'https://docs.github.com/en/free-pro-team@latest/rest/reference/'
#export
class GhApi(_GhObj):
def __init__(self, owner=None, repo=None, token=None, debug=None, limit_cb=None, **kwargs):
self.headers = { 'Accept': 'application/vnd.github.v3+json' }
token = token or os.getenv('GITHUB_TOKEN', None)
if token: self.headers['Authorization'] = 'token ' + token
if owner: kwargs['owner'] = owner
if repo: kwargs['repo' ] = repo
funcs_ = L(funcs).starmap(_GhVerb, client=self, kwargs=kwargs)
self.func_dict = {f'{o.path}:{o.verb.upper()}':o for o in funcs_}
self.groups = {k.replace('-','_'):_GhVerbGroup(k,v) for k,v in groupby(funcs_, 'tag').items()}
self.debug,self.limit_cb,self.limit_rem = debug,limit_cb,5000
def __call__(self, path:str, verb:str=None, headers:dict=None, route:dict=None, query:dict=None, data=None):
"Call a fully specified `path` using HTTP `verb`, passing arguments to `fastcore.core.urlsend`"
if verb is None: verb = 'POST' if data else 'GET'
headers = {**self.headers,**(headers or {})}
if path[:7] not in ('http://','https:/'): path = GH_HOST+path
res,self.recv_hdrs = urlsend(path, verb, headers=headers or None, debug=self.debug, return_headers=True,
route=route or None, query=query or None, data=data or None)
if 'X-RateLimit-Remaining' in self.recv_hdrs:
newlim = self.recv_hdrs['X-RateLimit-Remaining']
if self.limit_cb is not None and newlim != self.limit_rem:
self.limit_cb(int(newlim),int(self.recv_hdrs['X-RateLimit-Limit']))
self.limit_rem = newlim
return dict2obj(res)
def __dir__(self): return super().__dir__() + list(self.groups)
def _repr_markdown_(self): return "\n".join(f'- [{o}]({_docroot+o})' for o in sorted(self.groups))
def __getattr__(self,k): return self.groups[k] if 'groups' in vars(self) and k in self.groups else stop(AttributeError(k))
def __getitem__(self, k):
"Lookup and call an endpoint by path and verb (which defaults to 'GET')"
a,b = k if isinstance(k,tuple) else (k,'GET')
return self.func_dict[f'{a}:{b.upper()}']
def full_docs(self):
return '\n'.join(f'## {gn}\n\n{group._repr_markdown_()}\n' for gn,group in sorted(self.groups.items()))
#hide
token = os.environ['GITHUB_TOKEN']
show_doc(GhApi.__call__)
GhApi.__call__
[source]
GhApi.call
(path
:str
,verb
:str
=None
,headers
:dict
=None
,route
:dict
=None
,query
:dict
=None
,data
=None
)
Call a fully specified path
using HTTP verb
, passing arguments to fastcore.core.urlsend
api = GhApi()
You can call a GhApi
object as a function, passing in the path to the endpoint, the HTTP verb, and any route, query parameter, or post data parameters as required.
api('/repos/{owner}/{repo}/git/ref/{ref}', 'GET', route=dict(
owner='fastai', repo='ghapi-test', ref='heads/master'))
show_doc(GhApi.__getitem__)
GhApi.__getitem__
[source]
GhApi.getitem
(k
)
Lookup and call an endpoint by path and verb (which defaults to 'GET')
You can access endpoints by indexing into the object. When using the API this way, you do not need to specify what type of parameter (route, query, or post data) is being used. This is, therefore, the same call as above:
api['/repos/{owner}/{repo}/git/ref/{ref}'](owner='fastai', repo='ghapi-test', ref='heads/master')
GitHub has various rate limits for their API. After each call, the response includes information about how many requests are remaining in the hourly quota. If you'd like to add alerts, or indications showing current quota usage, you can register a callback with GhApi
by passing a callable to the limit_cb
parameter. This callback will be called whenever the amount of quota used changes. It will be called with two arguments: the new quota remaining, and the total hourly quota.
def _f(rem,quota): print(f"Quota remaining: {rem} of {quota}")
api = GhApi(limit_cb=_f)
api['/repos/{owner}/{repo}/git/ref/{ref}'](owner='fastai', repo='ghapi-test', ref='heads/master').ref
Quota remaining: 4997 of 5000
'refs/heads/master'
You can always get the remaining quota from the limit_rem
attribute:
api.limit_rem
'4997'
Instead of passing a path to GhApi
, you will more often use the operation methods provided in the API's operation groups, which include documentation, signatures, and auto-complete.
If you provide owner
and/or repo
to the constructor, they will be automatically inserted into any calls which use them (except when calling GhApi
as a function). You can also pass any other arbitrary keyword arguments you like to have them used as defaults for any relevant calls.
You must include a GitHub API token if you need to access any authenticated endpoints. If don't pass the token
param, then your GITHUB_TOKEN
environment variable will be used, if available.
api = GhApi(owner='fastai', repo='ghapi-test', token=token)
The following groups of endpoints are provided, which you can list at any time along with a link to documentation for all endpoints in that group, by displaying the GhApi
object:
api
The GitHub API's endpoint names generally start with a verb like "get", "list", "delete", "create", etc, followed _
, then by a noun such as "ref", "webhook", "issue", etc.
Each endpoint has a different signature, which you can see by using Shift-Tab in Jupyter, or by just printing the endpoint object (which also shows a link to the GitHub docs):
print(api.repos.create_webhook)
repos.create_webhook(name: str = None, config: dict = None, events: list = ['push'], active: bool = True) https://docs.github.com/rest/reference/repos#create-a-repository-webhook
Displaying an endpoint object in Jupyter also provides a formatted summary and link to the official GitHub documentation:
api.repos.create_webhook
repos.create_webhook(name, config, events, active): Create a repository webhook
Endpoint objects are called using standard Python method syntax:
ref = api.git.get_ref('heads/master')
test_eq(ref.object.type, 'commit')
Information about the endpoint are available as attributes:
api.git.get_ref.path,api.git.get_ref.verb
('/repos/fastai/ghapi-test/git/ref/{ref}', 'get')
You can get a list of all endpoints available in a group, along with a link to documentation for each, by viewing the group:
api.git
For "list" endpoints, the noun will be a plural form, e.g.:
#hide
for hook in api.repos.list_webhooks(): api.repos.delete_webhook(hook.id)
hooks = api.repos.list_webhooks()
test_eq(len(hooks), 0)
You can pass dicts, lists, etc. directly, where they are required for GitHub API endpoints:
url = 'https://example.com'
cfg = dict(url=url, content_type='json', secret='XXX')
hook = api.repos.create_webhook(config=cfg, events=['ping'])
test_eq(hook.config.url, url)
Let's confirm that our new webhook has been created:
hooks = api.repos.list_webhooks()
test_eq(len(hooks), 1)
test_eq(hooks[0].events, ['ping'])
Finally, we can delete our new webhook:
api.repos.delete_webhook(hooks[0].id)
#export
def date2gh(dt:datetime)->str:
"Convert `dt` (which is assumed to be in UTC time zone) to a format suitable for GitHub API operations"
return f'{dt.replace(microsecond=0).isoformat()}Z'
The GitHub API assumes that dates will be in a specific string format. date2gh
converts Python standard datetime
objects to that format. For instance, to find issues opened in the 'fastcore' repo in the last 4 weeks:
dt = date2gh(datetime.utcnow() - timedelta(weeks=4))
issues = GhApi('fastai').issues.list_for_repo(repo='fastcore', since=dt)
len(issues)
7
#export
def gh2date(dtstr:str)->datetime:
"Convert date string `dtstr` received from a GitHub API operation to a UTC `datetime`"
return datetime.fromisoformat(dtstr.replace('Z', ''))
created = issues[0].created_at
print(created, '->', gh2date(created))
2020-11-25T09:33:08Z -> 2020-11-25 09:33:08
GitHub's preview API functionality requires a special header to be passed to enable it. This is added automatically for you.
#export
def print_summary(req:Request):
"Print `Request.summary` with the token (if any) removed"
pprint(req.summary('Authorization'))
You can set the debug
attribute to any callable to intercept all requests, for instance to print Request.summary
. print_summary
is provided for this purpose. Using this, we can see the preview header that is added for preview functionality, e.g.
api.debug=print_summary
api.codes_of_conduct.get_all_codes_of_conduct()[0]
api.debug=None
{'data': None, 'full_url': 'https://api.github.com/codes_of_conduct', 'headers': {'Accept': 'application/vnd.github.scarlet-witch-preview+json'}, 'method': 'GET'}
There are some multi-step processes in the GitHub API that GhApi
provide convenient wrappers for. The methods currently available are shown below; do not hesitate to create an issue or pull request if there are other processes that you'd like to see supported better.
#export
@patch
def delete_release(self:GhApi, release):
"Delete a release and its associated tag"
self.repos.delete_release(release.id)
self.git.delete_ref(f'tags/{release.tag_name}')
#hide
for rel in api.repos.list_releases(): api.delete_release(rel)
#export
@patch
def upload_file(self:GhApi, rel, fn):
"Upload `fn` to endpoint for release `rel`"
fn = Path(fn)
url = rel.upload_url.replace('{?name,label}','')
mime = mimetypes.guess_type(fn, False)[0] or 'application/octet-stream'
return self(url, 'POST', headers={'Content-Type':mime}, query = {'name':fn.name}, data=fn.read_bytes())
#export
@patch
def create_release(self:GhApi, tag_name, branch='master', name=None, body='',
draft=False, prerelease=False, files=None):
"Wrapper for `GhApi.repos.create_release` which also uploads `files`"
if name is None: name = 'v'+tag_name
rel = self.repos.create_release(tag_name, target_commitish=branch, name=name, body=body,
draft=draft, prerelease=prerelease)
for file in listify(files): self.upload_file(rel, file)
return rel
Creating a release and attaching files to it is normally a multi-stage process, so create_release
wraps this up for you. It takes the same arguments as repos.create_release
, along with files
, which can contain a single file name, or a list of file names to upload to your release:
rel = api.create_release('0.0.1', files=['docs/index.html'])
test_eq(rel.name, 'v0.0.1')
rels = api.repos.list_releases()
test_eq(len(rels), 1)
We can check that our file has been uploaded; GitHub refers to them as "assets":
assets = api.repos.list_release_assets(rels[0].id)
test_eq(assets[0].name, 'index.html')
test_eq(assets[0].content_type, 'text/html')
show_doc(GhApi.delete_release)
#export
@patch
def list_tags(self:GhApi, prefix:str=''):
"List all tags, optionally filtered to those starting with `prefix`"
return self.git.list_matching_refs(f'tags/{prefix}')
With no prefix
, all tags are listed.
test_eq(len(api.list_tags()), 1)
Using the full tag name will return just that tag.
test_eq(len(api.list_tags(rel.tag_name)), 1)
#export
@patch
def list_branches(self:GhApi, prefix:str=''):
"List all branches, optionally filtered to those starting with `prefix`"
return self.git.list_matching_refs(f'heads/{prefix}')
Branches can be listed in the exactly the same way as tags.
test_eq(len(api.list_branches('master')), 1)
We can delete our release and confirm that it is removed:
api.delete_release(rels[0])
test_eq(len(api.repos.list_releases()), 0)
#export
# See https://stackoverflow.com/questions/9765453
EMPTY_TREE_SHA = '4b825dc642cb6eb9a060e54bf8d69288fbee4904'
#export
@patch
def create_branch_empty(self:GhApi, branch):
c = self.git.create_commit(f'create {branch}', EMPTY_TREE_SHA)
return self.git.create_ref(f'refs/heads/{branch}', c.sha)
ref = api.create_branch_empty("testme")
test_eq(len(api.list_branches('testme')), 1)
#export
@patch
def delete_tag(self:GhApi, tag:str):
"Delete a tag"
return self.git.delete_ref(f'tags/{tag}')
#export
@patch
def delete_branch(self:GhApi, branch:str):
"Delete a branch"
return self.git.delete_ref(f'heads/{branch}')
api.delete_branch('testme')
test_eq(len(api.list_branches('testme')), 0)
#export
@patch
def get_branch(self:GhApi, branch=None):
branch = branch or self.repos.get().default_branch
return self.list_branches(branch)[0]
#export
@patch
def list_files(self:GhApi, branch=None):
ref = self.get_branch(branch)
res = self.git.get_tree(ref.object.sha).tree
return {o.path:o for o in res}
files = api.list_files()
files['README.md']
#export
@patch
def get_content(self:GhApi, path):
res = self.repos.get_content(path)
return base64.b64decode(res.content)
readme = api.get_content('README.md').decode()
assert 'ghapi' in readme
#export
@patch
def update_contents(self:GhApi, path, message=None, content=None,
sha=None, branch=None, committer=None, author=None):
if sha is None: sha = api.list_files()[path].sha
if not isinstance(content,bytes): content = content.encode()
content = base64.b64encode(content).decode()
return api.repos.create_or_update_file_contents(path, message, content=content,
sha=sha, branch=branch, committer=committer, author=author)
res = api.update_contents('README.md', "Update README", content=readme+"foobar")
res.content.size
72
readme = api.get_content('README.md').decode()
assert 'foobar' in readme
api.update_contents('README.md', "Revert README", content=readme[:-6]);
#export
@patch
def enable_pages(self:GhApi, branch=None, path="/"):
"Enable or update pages for a repo to point to a `branch` and `path`."
if path not in ('/docs','/'): raise Exception("path not in ('/docs','/')")
r = self.repos.get()
branch = branch or r.default_branch
source = {"branch": branch, "path": path}
if r.has_pages: return self.repos.update_information_about_pages_site(source=source)
if len(self.list_branches(branch))==0: self.create_branch_empty(branch)
return self.repos.create_pages_site(source=source)
branch
is set to the default branch if None
. path
must be /docs
or /
.
res = api.enable_pages(branch='new-branch', path='/')
test_eq(res.source.branch, 'new-branch')
test_eq(res.source.path, '/')
api.repos.delete_pages_site()
api.delete_branch('new-branch')
#hide
from nbdev.export import notebook2script
notebook2script()
Converted 00_core.ipynb. Converted 01_actions.ipynb. Converted 02_auth.ipynb. Converted 03_page.ipynb. Converted 10_cli.ipynb. Converted 50_fullapi.ipynb. Converted 90_build_lib.ipynb. Converted event.ipynb. Converted index.ipynb. Converted tutorial_actions.ipynb.