Created
January 30, 2011 18:08
-
-
Save mrts/803066 to your computer and use it in GitHub Desktop.
GitHub API with descriptors
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# coding: utf-8 | |
""" | |
To my great disappointment, Dustin Sallings's otherwise excellent py-github | |
does not work with repositories that have / in branch names, as this breaks | |
the XML-based parser. Unfortunately, many projects use / for simulating | |
directory structure. See e.g. | |
http://github.com/api/v2/xml/repos/show/django/django/branches | |
for the error. | |
What follows is a comprehensive interface to the GitHub JSON API that does not | |
suffer from the above limitation and strives to be even more Pythonic than its | |
predecessor. | |
Another notable addition is caching. By default, everything is cached to avoid | |
unneccessary roundtrips to GitHub (otherwise one could stumble upon GitHub API | |
call limits during intensive usage besides wasting bandwidth pointlessly). Of | |
course, the cache can be both cleared and disabled if required. | |
Currently, the module is user-centered, i.e. general searching is not | |
implemented (as I haven't encountered a useful use-case for that yet). Please | |
let me know if you would want to see the search API wrapped as well (or just | |
fork on GitHub, implement it and send me a pull request). | |
Invoke `python github.py` to run the following usage examples. | |
TODO: | |
* cache lives with the main user object and can be cleared. | |
* review py-github for good ideas | |
>>> u = GitHubUser('691a79ee3da706') | |
Traceback (most recent call last): | |
... | |
GitHubUser.DoesNotExist: the user '691a79ee3da706' does not exist. | |
>>> u = GitHubUser('mrts') | |
>>> u.details | |
<<UserDetails for mrts>> | |
>>> u.details.name.encode('ascii', 'xmlcharrefreplace') | |
'Mart Sõmermaa' | |
>>> u.details.email | |
u'mrts.pydev at gmail dot com' | |
>>> u.repositories | |
{u'cve_tracker': <<Repository cve_tracker>>, u'qparams': <<Repository qparams>>, u'django-commands': <<Repository django-commands>>, u'dotfiles': <<Repository dotfiles>>, u'OpenSC': <<Repository OpenSC>>, u'django': <<Repository django>>, u'plugit': <<Repository plugit>>} | |
>>> u.repositories['django'].fork | |
True | |
>>> u.repositories['django'].branches | |
set([u'soc2009/i18n-improvements', u'soc2009/multidb', u'formset_refactor', u'soc2009/model-validation', u'ticket11967-1.1.X', u'raw_sql_override', u'1.0.X', u'1.1.X-mergequeue', u'ticket12769', u'releases/1.0.X', u'ticket12780-1.1.X', u'ticket12780', u'master', u'soc2009/http-wsgi-improvements', u'ticket6422-1.1.X', u'ticket7028-1.1.X', u'soc2009/admin-ui', u'soc2009/test-improvements', u'ticket7028', u'1.1.X']) | |
""" | |
import json, httplib, warnings | |
from contextlib import closing | |
class GitHubProxy(object): | |
HOST = 'github.com' | |
URL_BASE = '/api/v2/json/' | |
def __init__(self): | |
self._cache = {} | |
def _fetch(self, **kwargs): | |
assert kwargs, "Provide at least one argument" | |
full_url = (self.URL_BASE + self.path) % kwargs | |
with closing(httplib.HTTPConnection(self.HOST)) as conn: | |
conn.request("GET", full_url) | |
resp = conn.getresponse() | |
if resp.status == httplib.OK: | |
return json.loads(resp.read()) | |
elif resp.status == httplib.NOT_FOUND: | |
return False | |
else: | |
raise httplib.HTTPException("Unexpected response status: %s" % | |
resp.status) | |
def __set__(self, obj, value): | |
raise AttributeError("Setting properties is not supported") | |
class NickBasedDescriptor(GitHubProxy): | |
def __get__(self, obj, cls): | |
if obj.nick not in self._cache: | |
resp = self._fetch(user=obj.nick) | |
self._cache[obj.nick] = self._handle_response(resp) | |
return self._cache[obj.nick] | |
class UserDetailsDescriptor(NickBasedDescriptor): | |
path = "user/show/%(user)s" | |
def _handle_response(self, resp): | |
return (UserDetails(resp['user']) if resp else False) | |
class RepositoryDescriptor(NickBasedDescriptor): | |
path = "repos/show/%(user)s" | |
def _handle_response(self, resp): | |
if resp: | |
return dict((repo['name'], Repository(repo)) for repo in | |
resp['repositories']) | |
else: | |
return {} | |
class BranchDescriptor(GitHubProxy): | |
path = "repos/show/%(user)s/%(repo)s/branches" | |
def __get__(self, obj, cls): | |
if obj.name not in self._cache: | |
resp = self._fetch(user=obj.owner, repo=obj.name) | |
self._cache[obj.name] = self._handle_response(resp) | |
return self._cache[obj.name] | |
def _handle_response(self, resp): | |
if resp: | |
return set(resp['branches'].keys()) | |
else: | |
return set() | |
class DictBased(object): | |
def __init__(self, dct): | |
self.__dict__.update(dct) | |
class UserDetails(DictBased): | |
login = None | |
name = '' | |
email = '' | |
def __repr__(self): | |
return u'<<UserDetails for %s>>' % self.login | |
class Repository(DictBased): | |
branches = BranchDescriptor() | |
def __repr__(self): | |
return u'<<Repository %s>>' % self.name | |
class GitHubUser(object): | |
details = UserDetailsDescriptor() | |
repositories = RepositoryDescriptor() | |
def __init__(self, nick, token=None, secure=False, disable_cache=False): | |
self.nick = nick # TODO: escape nick? pointlessish... | |
self.token = token | |
self.secure = secure | |
if token and not secure: | |
warnings.warn("Authentication token should not be transmitted " | |
"in clear, enabling secure mode (HTTPS). " | |
"Always use 'secure=True' when using the token.") | |
self.secure = True | |
self.disable_cache = disable_cache | |
@property | |
def exists(self): | |
return bool(self.details) | |
def __repr__(self): | |
return u'<<GitHubUser %s>>' % self.nick | |
if __name__ == '__main__': | |
import doctest | |
doctest.testmod() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment