{ "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 datetime import datetime,timedelta\n", "from pprint import pprint" ] }, { "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 = \"https://api.github.com\"\n", "_DOC_URL = 'https://docs.github.com/'" ] }, { "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 = 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, debug=None, limit_cb=None, **kwargs):\n", " self.headers = { 'Accept': 'application/vnd.github.v3+json' }\n", " token = token or os.getenv('GITHUB_TOKEN', None)\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", "\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 path[:7] not in ('http://','https:/'): path = GH_HOST+path\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})' 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": [
"