# 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",
}