{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#default_exp core" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# GhApi details\n", "\n", "> Detailed information on the GhApi API" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "from fastcore.utils import *\n", "from fastcore.foundation import *\n", "from fastcore.meta import *\n", "from ghapi.metadata import funcs\n", "\n", "import mimetypes,base64\n", "from inspect import signature,Parameter,Signature\n", "from urllib.request import Request\n", "from urllib.error import HTTPError\n", "from urllib.parse import quote\n", "from datetime import datetime,timedelta\n", "from pprint import pprint\n", "import os" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#hide\n", "from nbdev import *" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "GH_HOST = os.getenv('GH_HOST', \"https://api.github.com\")\n", "_DOC_URL = 'https://docs.github.com/'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "You can set an environment variable named `GH_HOST` to override the default of `https://api.github.com` incase you are running [GitHub Enterprise](https://github.com/enterprise)(GHE). However, this library has not been tested on GHE, so proceed at your own risk." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "def _preview_hdr(preview): return {'Accept': f'application/vnd.github.{preview}-preview+json'} if preview else {}\n", "\n", "def _mk_param(nm, **kwargs): return Parameter(nm, kind=Parameter.POSITIONAL_OR_KEYWORD, **kwargs)\n", "def _mk_sig_detls(o):\n", " res = {}\n", " if o[0]!=object: res['annotation']=o[0]\n", " res['default'] = o[1] if len(o)>1 else None\n", " return res\n", "def _mk_sig(req_args, opt_args, anno_args):\n", " params = [_mk_param(k) for k in req_args]\n", " params += [_mk_param(k, default=v) for k,v in opt_args.items()]\n", " params += [_mk_param(k, **_mk_sig_detls(v)) for k,v in anno_args.items()]\n", " return Signature(params)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "class _GhObj: pass" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "class _GhVerb(_GhObj):\n", " __slots__ = 'path,verb,tag,name,summary,url,route_ps,params,data,preview,client,__doc__'.split(',')\n", " def __init__(self, path, verb, oper, summary, url, params, data, preview, client, kwargs):\n", " tag,*name = oper.split('/')\n", " name = '__'.join(name)\n", " name = name.replace('-','_')\n", " path,_,_ = partial_format(path, **kwargs)\n", " route_ps = stringfmt_names(path)\n", " __doc__ = summary\n", " data = {o[0]:o[1:] for o in data}\n", " store_attr()\n", "\n", " def __call__(self, *args, headers=None, **kwargs):\n", " headers = {**_preview_hdr(self.preview),**(headers or {})}\n", " d = list(self.data)\n", " flds = [o for o in self.route_ps+self.params+d if o not in kwargs]\n", " for a,b in zip(args,flds): kwargs[b]=a\n", " kwargs = {k:v for k,v in kwargs.items() if v is not None}\n", " route_p,query_p,data_p = [{p:kwargs[p] for p in o if p in kwargs}\n", " for o in (self.route_ps,self.params,d)]\n", " return self.client(self.path, self.verb, headers=headers, route=route_p, query=query_p, data=data_p)\n", "\n", " def __str__(self): return f'{self.tag}.{self.name}{signature(self)}\\n{self.doc_url}'\n", " @property\n", " def __signature__(self): return _mk_sig(self.route_ps, dict.fromkeys(self.params), self.data)\n", " __call__.__signature__ = __signature__\n", " @property\n", " def doc_url(self): return _DOC_URL + self.url.replace(\" \",\"_\")\n", "\n", " def _repr_markdown_(self):\n", " params = ', '.join(self.route_ps+self.params+list(self.data))\n", " return f'[{self.tag}.{self.name}]({self.doc_url})({params}): *{self.summary}*'\n", " __repr__ = _repr_markdown_" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "class _GhVerbGroup(_GhObj):\n", " def __init__(self, name, verbs):\n", " self.name,self.verbs = name,verbs\n", " for o in verbs: setattr(self, o.name, o)\n", " def __str__(self): return \"\\n\".join(str(v) for v in self.verbs)\n", " def _repr_markdown_(self): return \"\\n\".join(f'- {v._repr_markdown_()}' for v in self.verbs)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "_docroot = 'https://docs.github.com/en/free-pro-team@latest/rest/reference/'" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## GhApi -" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#export\n", "class GhApi(_GhObj):\n", " def __init__(self, owner=None, repo=None, token=None, jwt_token=None, debug=None, limit_cb=None, gh_host=None, **kwargs):\n", " self.headers = { 'Accept': 'application/vnd.github.v3+json' }\n", " token = token or os.getenv('GITHUB_TOKEN', None)\n", " jwt_token = jwt_token or os.getenv('GITHUB_JWT_TOKEN', None)\n", " if jwt_token: self.headers['Authorization'] = 'Bearer ' + jwt_token\n", " if token: self.headers['Authorization'] = 'token ' + token\n", " if owner: kwargs['owner'] = owner\n", " if repo: kwargs['repo' ] = repo\n", " funcs_ = L(funcs).starmap(_GhVerb, client=self, kwargs=kwargs)\n", " self.func_dict = {f'{o.path}:{o.verb.upper()}':o for o in funcs_}\n", " self.groups = {k.replace('-','_'):_GhVerbGroup(k,v) for k,v in groupby(funcs_, 'tag').items()}\n", " self.debug,self.limit_cb,self.limit_rem = debug,limit_cb,5000\n", " self.gh_host = gh_host or GH_HOST\n", "\n", " def __call__(self, path:str, verb:str=None, headers:dict=None, route:dict=None, query:dict=None, data=None):\n", " \"Call a fully specified `path` using HTTP `verb`, passing arguments to `fastcore.core.urlsend`\"\n", " if verb is None: verb = 'POST' if data else 'GET'\n", " headers = {**self.headers,**(headers or {})}\n", " if not path.startswith(('http://', 'https://')):\n", " path = self.gh_host + path\n", " if route:\n", " for k,v in route.items(): route[k] = quote(str(route[k]))\n", " res,self.recv_hdrs = urlsend(path, verb, headers=headers or None, debug=self.debug, return_headers=True,\n", " route=route or None, query=query or None, data=data or None)\n", " if 'X-RateLimit-Remaining' in self.recv_hdrs:\n", " newlim = self.recv_hdrs['X-RateLimit-Remaining']\n", " if self.limit_cb is not None and newlim != self.limit_rem:\n", " self.limit_cb(int(newlim),int(self.recv_hdrs['X-RateLimit-Limit']))\n", " self.limit_rem = newlim\n", "\n", " return dict2obj(res)\n", "\n", " def __dir__(self): return super().__dir__() + list(self.groups)\n", " def _repr_markdown_(self): return \"\\n\".join(f\"- [{o}]({_docroot + o.replace('_', '-')})\" for o in sorted(self.groups))\n", " def __getattr__(self,k): return self.groups[k] if 'groups' in vars(self) and k in self.groups else stop(AttributeError(k))\n", "\n", " def __getitem__(self, k):\n", " \"Lookup and call an endpoint by path and verb (which defaults to 'GET')\"\n", " a,b = k if isinstance(k,tuple) else (k,'GET')\n", " return self.func_dict[f'{a}:{b.upper()}']\n", "\n", " def full_docs(self):\n", " return '\\n'.join(f'## {gn}\\n\\n{group._repr_markdown_()}\\n' for gn,group in sorted(self.groups.items()))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "#hide\n", "token = os.environ['GITHUB_TOKEN']" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Access by path" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [ { "data": { "text/markdown": [ "
GhApi.__call__
[source]GhApi.__call__
(**`path`**:`str`, **`verb`**:`str`=*`None`*, **`headers`**:`dict`=*`None`*, **`route`**:`dict`=*`None`*, **`query`**:`dict`=*`None`*, **`data`**=*`None`*)\n",
"\n",
"Call a fully specified `path` using HTTP `verb`, passing arguments to `fastcore.core.urlsend`"
],
"text/plain": [
"GhApi.__getitem__
[source]GhApi.__getitem__
(**`k`**)\n",
"\n",
"Lookup and call an endpoint by path and verb (which defaults to 'GET')"
],
"text/plain": [
"GhApi.delete_release
[source]GhApi.delete_release
(**`release`**)\n",
"\n",
"Delete a release and its associated tag"
],
"text/plain": [
"