Source code for github2.request

# Copyright (C) 2009-2012 Adam Vandenberg <flangy@gmail.com>
#                         Asheesh Laroia <asheesh@openhatch.org>
#                         Ask Solem <askh@modwheel.net>
#                         Chris Vale <crispywalrus@gmail.com>
#                         Daniel Greenfeld <pydanny@gmail.com>
#                         Evan Broder <broder@mit.edu>
#                         James Rowe <jnrowe@gmail.com>
#                         Jeremy Dunck <jdunck@gmail.com>
#                         Josh Weinberg <daemoncollector@gmail.com>
#                         Mark Paschal <markpasc@markpasc.org>
#                         Maximillian Dornseif <m.dornseif@hudora.de>
#                         Michael Basnight <mbasnight@gmail.com>
#                         Patryk Zawadzki <patrys@pld-linux.org>
#                         Rick Harris <rick.harris@rackspace.com>
#                         Sameer Al-Sakran <sameer@whitelabellabs.com>
#                         Vincent Driessen <vincent@datafox.nl>
#                         modocache <modocache@gmail.com>
#
# This file is part of python-github2, and is made available under the 3-clause
# BSD license.  See LICENSE for the full details.

import datetime
import logging
import re
import sys
import time

try:
    # For Python 3
    from http.client import responses
except ImportError:  # For Python 2.5-2.7
    try:
        from httplib import responses  # NOQA
    except ImportError:  # For Python 2.4
        from BaseHTTPServer import BaseHTTPRequestHandler as _BHRH
        responses = dict([(k, v[0]) for k, v in _BHRH.responses.items()])  # NOQA
try:
    import json as simplejson  # For Python 2.6+
except ImportError:
    import simplejson  # NOQA
from os import (getenv, path)
try:
    # For Python 3
    from urllib.parse import (parse_qs, quote, urlencode, urlsplit, urlunsplit)
except ImportError:
    from urlparse import (urlsplit, urlunsplit)  # NOQA
    try:
        from urlparse import parse_qs  # NOQA
    except ImportError:
        from cgi import parse_qs  # NOQA
    from urllib import urlencode, quote  # NOQA

import httplib2


#: Hostname for API access
DEFAULT_GITHUB_URL = "https://github.com"

#: Logger for requests module
LOGGER = logging.getLogger('github2.request')

# Fetch actual path for httplib2's default cert bundle, for distributions that
# symlink their system certs
_HTTPLIB2_BUNDLE = path.realpath(path.dirname(httplib2.CA_CERTS))
#: Whether github2 is using the system's certificates for SSL connections
SYSTEM_CERTS = not _HTTPLIB2_BUNDLE.startswith(path.dirname(httplib2.__file__))
CA_CERTS = None
#: Whether github2 is using the cert's from the file given in $CURL_CA_BUNDLE
CURL_CERTS = False
if not SYSTEM_CERTS and sys.platform.startswith('linux'):
    for cert_file in ['/etc/ssl/certs/ca-certificates.crt',
                      '/etc/pki/tls/certs/ca-bundle.crt']:
        if path.exists(cert_file):
            CA_CERTS = cert_file
            SYSTEM_CERTS = True
            break
elif not SYSTEM_CERTS and sys.platform.startswith('freebsd'):
    if path.exists('/usr/local/share/certs/ca-root-nss.crt'):
        CA_CERTS = '/usr/local/share/certs/ca-root-nss.crt'
        SYSTEM_CERTS = True
elif path.exists(getenv('CURL_CA_BUNDLE', '')):
    CA_CERTS = getenv('CURL_CA_BUNDLE')
    CURL_CERTS = True
if not SYSTEM_CERTS and not CURL_CERTS:
    CA_CERTS = path.join(path.dirname(path.abspath(__file__)),
                         "DigiCert_High_Assurance_EV_Root_CA.crt")


# Common missing entries from the HTTP status code dict, basically anything
# GitHub reports that isn't basic HTTP/1.1.
responses[422] = 'Unprocessable Entity'


def charset_from_headers(headers):
    """Parse charset from headers.

    :param httplib2.Response headers: Request headers
    :return: Defined encoding, or default to ASCII

    """
    match = re.search("charset=([^ ;]+)", headers.get('content-type', ""))
    if match:
        charset = match.groups()[0]
    else:
        charset = "ascii"
    return charset


[docs]class GithubError(Exception): """An error occurred when making a request to the GitHub API."""
class HttpError(RuntimeError): """A HTTP error occured when making a request to the GitHub API.""" def __init__(self, message, content, code): """Create a HttpError exception. :param str message: Exception string :param str content: Full content of HTTP request :param int code: HTTP status code """ self.args = (message, content, code) self.message = message self.content = content self.code = code if code in responses: self.code_reason = responses[code] else: self.code_reason = "<unknown status code>" LOGGER.warning('Unknown HTTP status %r, please file an issue', code)
[docs]class GithubRequest(object): """Make an API request. :see: :class:`github2.client.Github` """ url_format = "%(github_url)s/api/%(api_version)s/%(api_format)s" api_version = "v2" api_format = "json" GithubError = GithubError def __init__(self, username=None, api_token=None, url_prefix=None, requests_per_second=None, access_token=None, cache=None, proxy_host=None, proxy_port=None, github_url=None): self.username = username self.api_token = api_token self.access_token = access_token self.url_prefix = url_prefix if github_url is None: self.github_url = DEFAULT_GITHUB_URL else: self.github_url = github_url if requests_per_second is None: self.delay = 0 else: self.delay = 1.0 / requests_per_second self.last_request = datetime.datetime(1900, 1, 1) if not self.url_prefix: self.url_prefix = self.url_format % { "github_url": self.github_url, "api_version": self.api_version, "api_format": self.api_format, } if proxy_host is None: self._http = httplib2.Http(cache=cache) else: proxy_info = httplib2.ProxyInfo(httplib2.socks.PROXY_TYPE_HTTP, proxy_host, proxy_port) self._http = httplib2.Http(proxy_info=proxy_info, cache=cache) self._http.ca_certs = CA_CERTS if SYSTEM_CERTS: LOGGER.info('Using system certificates in %r', CA_CERTS) elif CURL_CERTS: LOGGER.info("Using cURL's certificates in %r", CA_CERTS) else: LOGGER.warning('Using bundled certificate for HTTPS connections') def encode_authentication_data(self, extra_post_data): post_data = [] if self.access_token: post_data.append(("access_token", self.access_token)) elif self.username and self.api_token: post_data.append(("login", self.username)) post_data.append(("token", self.api_token)) for key, value in extra_post_data.items(): if isinstance(value, list): for elem in value: post_data.append((key, elem)) else: post_data.append((key, value)) return urlencode(post_data) def get(self, *path_components): path_components = filter(None, path_components) return self.make_request("/".join(path_components)) def post(self, *path_components, **extra_post_data): path_components = filter(None, path_components) return self.make_request("/".join(path_components), extra_post_data, method="POST") def put(self, *path_components, **extra_post_data): path_components = filter(None, path_components) return self.make_request("/".join(path_components), extra_post_data, method="PUT") def delete(self, *path_components, **extra_post_data): path_components = filter(None, path_components) return self.make_request("/".join(path_components), extra_post_data, method="DELETE") def make_request(self, path, extra_post_data=None, method="GET"): if self.delay: since_last = (datetime.datetime.utcnow() - self.last_request) if since_last.days == 0 and since_last.seconds < self.delay: duration = self.delay - since_last.seconds LOGGER.warning("delaying API call %g second(s)", duration) time.sleep(duration) extra_post_data = extra_post_data or {} url = "/".join([self.url_prefix, quote(path)]) result = self.raw_request(url, extra_post_data, method=method) if self.delay: self.last_request = datetime.datetime.utcnow() return result def raw_request(self, url, extra_post_data, method="GET"): scheme, netloc, path, query, fragment = urlsplit(url) post_data = None headers = self.http_headers method = method.upper() if extra_post_data or method == "POST": post_data = self.encode_authentication_data(extra_post_data) headers["Content-Length"] = str(len(post_data)) else: query = self.encode_authentication_data(parse_qs(query)) url = urlunsplit((scheme, netloc, path, query, fragment)) response, content = self._http.request(url, method, post_data, headers) if LOGGER.isEnabledFor(logging.DEBUG): logging.debug("URL: %r POST_DATA: %r RESPONSE_TEXT: %r", url, post_data, content) if response.status >= 400: raise HttpError("Unexpected response from github.com %d: %r" % (response.status, content), content, response.status) json = simplejson.loads(content.decode(charset_from_headers(response))) if json.get("error"): raise self.GithubError(json["error"][0]["error"]) return json @property def http_headers(self): return { "User-Agent": "pygithub2 v1", "Accept": "application/json", }