GitHub API wrapper

github.py

"""
GitHub API v3 Wrapper

Ref: https://developer.github.com/v3/git/

"""

import base64
import json
import requests

GITHUB_API_BASE = 'https://api.github.com'

class GitError(Exception):
    pass

class Transport(object):
    "Encapsulate HTTPS/JSON transport functions"

    def __init__(self, auth_user=None, auth_pass=None):
        self.__s = requests.Session()
        if auth_user:
            self.__auth = requests.auth.HTTPBasicAuth(auth_user, auth_pass)
        else:
            self.__auth = None

    def get(self, uri):
        uri = GITHUB_API_BASE + uri
        #print('GET: ' + uri)
        r = self.__s.get(uri,
            headers={
                'accept': 'application/vnd.github.v3+json',
            },
            auth=self.__auth)
        r.raise_for_status()
        return r.json()

    def post(self, uri, data):
        uri = GITHUB_API_BASE + uri
        data=json.dumps(data)
        #print('POST: ' + uri)
        #print(data)
        r = self.__s.post(uri,
            auth=self.__auth,
            headers={
                'content-type': 'application/json',
                'accept': 'application/vnd.github.v3+json',
            },
            data=data)
        r.raise_for_status()
        return r.json()

    def patch(self, uri, data):
        uri = GITHUB_API_BASE + uri
        #print('PATCH: ' + uri)
        r = self.__s.patch(uri,
            auth=self.__auth,
            headers={
                'content-type': 'application/json',
                'accept': 'application/vnd.github.v3+json',
            },
            data=json.dumps(data))
        r.raise_for_status()
        return r.json()

    def delete(self, uri):
        uri = GITHUB_API_BASE + uri
        #print('DELETE: ' + uri)
        r = self.__s.delete(uri,
            headers={
                'accept': 'application/vnd.github.v3+json',
            },
            auth=self.__auth)
        r.raise_for_status()


class GitObjectsCollection(object):
    def __init__(self, git, base_uri):
        self.git = git
        self._base_uri = base_uri

    def __getitem__(self, hash):
        try:
            r = self.git._transport.get(self._base_uri + '/' + hash)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code in [404, 409]:
                raise KeyError(hash)
        return self._parse(r)

    def _do_create(self, data):
        return self.git._transport.post(self._base_uri, data)

    def _create(self, data):
        return self._parse(self._do_create(data))

    def _parse(self, data):
        raise NotImplementedError()


class GitBlob(object):
    def __init__(self, hash, size, content):
        self.hash = hash
        self.size = size
        self.content = content

class GitBlobs(GitObjectsCollection):
    def __init__(self, git, base_uri):
        GitObjectsCollection.__init__(self, git, base_uri)

    def create(self, content):
        r = self._do_create({
            'content': content,
            'encoding': 'base64',
        })
        return GitBlob(r['sha'], len(content), content)

    def _parse(self, data):
        return GitBlob(data['sha'], data['size'], data['content'])


class GitTree(object):
    def __init__(self, hash, truncated, objects):
        self.hash = hash
        self.truncated = truncated
        self.objects = objects

class GitTrees(GitObjectsCollection):
    def __init__(self, git, base_uri):
        GitObjectsCollection.__init__(self, git, base_uri)

    def create(self, tree, base_tree=None):
        o = {
            'tree': tree,
        }
        if base_tree:
            o['base_tree'] = base_tree
        return self._create(o)

    def _parse(self, data):
        return GitTree(data['sha'], data.get('truncated', False), data['tree'])


class GitCommit(object):
    def __init__(self, hash, message, tree_ref, parent_refs, author, committer):
        self.hash = hash
        self.author = author
        self.committer = committer
        self.message = message
        self.tree_ref = tree_ref
        self.parent_refs = parent_refs

class GitCommits(GitObjectsCollection):
    def __init__(self, git, base_uri):
        GitObjectsCollection.__init__(self, git, base_uri)

    def create(self, message, tree_ref, parent_refs=None, author=None, committer=None):
        o = {
            'message': message,
            'tree': tree_ref,
        }
        if parent_refs:
            o['parents'] = parent_refs
        if author:
            o['author'] = author
        if committer:
            o['committer'] = committer
        return self._create(o)

    def _parse(self, data):
        return GitCommit(data['sha'],
            data['message'],
            data['tree']['sha'],
            [x['sha'] for x in data['parents']],
            data.get('author'),
            data.get('committer'))


class GitRef(object):
    def __init__(self, ref, hash, object_type):
        self.ref = ref
        self.hash = hash
        self.object_type = object_type
    def __str__(self):
        return '%s -> %s %s' % (self.ref, self.object_type, self.hash)

class GitRefs(GitObjectsCollection):
    def __init__(self, git, base_uri):
        GitObjectsCollection.__init__(self, git, base_uri)

    def list(self):
        try:
            r = self.git._transport.get(self._base_uri)
            return [self._parse(data) for data in r]
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 409:
                return []

    def create(self, ref, hash):
        return self._create({
            'ref': 'refs/' + ref,
            'sha': hash,
        })

    def update(self, ref, hash):
        r = self.git._transport.patch(self._base_uri + '/' + ref, {
            'sha': hash,
        })
        return self._parse(r)

    def delete(self, ref):
        self.git._transport.delete(self._base_uri + '/' + ref)

    def _parse(self, data):
        return GitRef(data['ref'][5:], data['object']['sha'], data['object']['type'])


class Git(object):
    def __init__(self, owner, repo, auth_user=None, auth_pass=None):
        self._transport = Transport(auth_user, auth_pass)
        base_uri = '/repos/%s/%s/git' % (owner, repo)
        self.blobs = GitBlobs(self, base_uri + '/blobs')
        self.trees = GitTrees(self, base_uri + '/trees')
        self.commits = GitCommits(self, base_uri + '/commits')
        self.refs = GitRefs(self, base_uri + '/refs')

# EOF