res = {}

if o[0]!=object: res['annotation']=o[0]

res['default'] = o[1] if len(o)>1 else None

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()]

__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 = '__'.join(name)

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

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}*'

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 __init__(self, owner=None, repo=None, token=None, jwt_token=None, debug=None, limit_cb=None, gh_host=None,

authenticate=True, **kwargs):

self.headers = { 'Accept': 'application/vnd.github.v3+json' }

if authenticate:

token = token or os.getenv('GITHUB_TOKEN', None)

jwt_token = jwt_token or os.getenv('GITHUB_JWT_TOKEN', None)

if jwt_token: self.headers['Authorization'] = 'Bearer ' + jwt_token

elif token: self.headers['Authorization'] = 'token ' + token

else: warn('Neither GITHUB_TOKEN nor GITHUB_JWT_TOKEN found: running as unauthenticated')

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

self.gh_host = gh_host or GH_HOST

def __call__(self, path:str, verb:str=None, headers:dict=None, route:dict=None, query:dict=None, data=None, timeout=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 not path.startswith(('http://', 'https://')):

path = self.gh_host + path

if route:

for k,v in route.items(): route[k] = quote(str(route[k]), safe='')

debug = self.debug if self.debug else print_summary if os.getenv('GHAPI_DEBUG') else None

res,self.recv_hdrs = urlsend(path, verb, headers=headers or None, decode=False, debug=debug, return_headers=True,

route=route or None, query=query or None, data=data or None, return_json=False, timeout=timeout)

ct = self.recv_hdrs.get('Content-Type', '')

if not any(t in ct for t in _binary_cts): res = res.decode()

if 'json' in ct: res = loads(res)

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) if isinstance(res, (dict, list)) else res

def __dir__(self): return super().__dir__() + list(self.groups)

def _repr_markdown_(self): return "\n".join(f"- [{o}]({_docroot + o.replace('_', '-')})" 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):

'Create a gist, optionally with images where each md img url will be placed with img upload urls.'

gist = self.gists.create(description, public=public, files={filename: {"content": content}})

if not img_paths: return gist

with tempfile.TemporaryDirectory() as clone_dir:

token = self.headers['Authorization'].split('token ')[1]

_run_subp(['git', 'clone', f'https://{token}@gist.github.com/{gist.id}.git', clone_dir])

clone_dir, img_paths = Path(clone_dir), L(img_paths).map(Path)

for o in img_paths: shutil.copy2(o, clone_dir/o.name)

_run_subp(['git', '-C', clone_dir, 'add', '.'])

_run_subp(['git', '-C', clone_dir, 'commit', '-m', 'Add images'])

_run_subp(['git', '-C', clone_dir, 'push'])

updated_gist = self.gists.get(gist.id)

img_urls = {o.name: updated_gist.files[o.name].raw_url for o in img_paths}

content = img_md_pat.sub(lambda m: f"![{m['alt']}]({img_urls.get(m['url'], m['url'])})", content)

"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'

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)

t = self.git.create_tree(base_tree=EMPTY_TREE_SHA, tree = [dict(

path='.dummy', content='ignore me', mode='100644', type='blob')])

c = self.git.create_commit(f'create {branch}', t.sha)

if not branch: branch = api.repos.get()['default_branch']

if not isinstance(content,bytes): content = content.encode()

content = base64.b64encode(content).decode()

kwargs = {'sha':sha} if sha else {}

return self.repos.create_or_update_file_contents(path, message, content=content,

if not branch: branch = api.repos.get()['default_branch']

if sha is None: sha = self.list_files()[path].sha

return self.repos.delete_file(path, message=message, sha=sha,

"Returns True if path matches include patterns (if any) and doesn't match any exclude pattern."

if include and not any(fnmatch.fnmatch(path, p) for p in listify(include)): return False

if exclude and any(fnmatch.fnmatch(path, p) for p in listify(exclude)): return False

"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)