kennethreitz/requests
1
0
Fork 0
mirror of https://github.com/kennethreitz/requests.git synced 2026-06-05 22:50:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add in a proper AuthManager instead of the list version that was being used.

Browse Source 
Added support for all Auth types that python supports
This commit is contained in:
digitalxero
2011-02-19 15:04:56 -05:00
parent a9e18e9106
commit edfdef2396
4 changed files with 158 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.rst
+1 -28
View File
@@ -38,7 +38,7 @@ HTTPS? Basic Authentication? ::
Uh oh, we're not authorized! Let's add authentication. ::
>>> conv_auth = ('requeststest', 'requeststest')
>>> conv_auth = requests.AuthObject('requeststest', 'requeststest')
>>> r = requests.get('https://convore.com/api/account/verify.json', auth=conv_auth)
>>> r.status_code
@@ -82,33 +82,6 @@ If CookieJar object is is passed in (cookies=...), the cookies will be sent with
>>> request.delete(url, params={}, headers={}, cookies=None, auth=None)
<request object>
**AsyncRequests:**
All request functions return a Response object (see below).
If a {filename: fileobject} dictionary is passed in (files=...), a multipart_encode upload will be performed.
If CookieJar object is is passed in (cookies=...), the cookies will be sent with the request.
GET Requests
>>> request.async.get(url, params={}, headers={}, cookies=None, auth=None)
<request object>
HEAD Requests
>>> request.async.head(url, params={}, headers={}, cookies=None, auth=None)
<request object>
PUT Requests
>>> request.async.put(url, data='', headers={}, files={}, cookies=None, auth=None)
<request object>
POST Requests
>>> request.async.post(url, data={}, headers={}, files={}, cookies=None, auth=None)
<request object>
DELETE Requests
>>> request.async.delete(url, params={}, headers={}, cookies=None, auth=None)
<request object>
**Responses:**
requests/async.py
+1 -1
View File
@@ -33,7 +33,7 @@ if not 'eventlet' in locals():
from .core import *
__all__ = ['Request', 'Response', 'request', 'get', 'head', 'post', 'put', 'delete', 'add_autoauth', 'AUTOAUTHS',
__all__ = ['Request', 'Response', 'request', 'get', 'head', 'post', 'put', 'delete', 'auth_manager', 'AuthObject',
'RequestException', 'AuthenticationError', 'URLRequired', 'InvalidMethod', 'HTTPError']
__title__ = 'requests'
__version__ = '0.0.1'
requests/core.py
+155 -70
View File
@@ -18,18 +18,15 @@ from urllib2 import HTTPError
from .packages.poster.encode import multipart_encode
from .packages.poster.streaminghttp import register_openers
__all__ = ['Request', 'Response', 'request', 'get', 'head', 'post', 'put', 'delete', 'add_autoauth', 'AUTOAUTHS',
__all__ = ['Request', 'Response', 'request', 'get', 'head', 'post', 'put', 'delete', 'auth_manager', 'AuthObject',
'RequestException', 'AuthenticationError', 'URLRequired', 'InvalidMethod', 'HTTPError']
__title__ = 'requests'
__version__ = '0.2.5'
__build__ = 0x000205
__author__ = 'Kenneth Reitz'
__author__ = 'Kenneth Reitz, Dj Gilcrease'
__license__ = 'ISC'
__copyright__ = 'Copyright 2011 Kenneth Reitz'
AUTOAUTHS = []
class _Request(urllib2.Request):
"""Hidden wrapper around the urllib2.Request object. Allows for manual
setting of HTTP methods.
@@ -69,6 +66,10 @@ class Request(object):
self.response = Response()
if isinstance(auth, (list, tuple)):
auth = AuthObject(*auth)
if not auth:
auth = auth_manager.get_auth(self.url)
self.auth = auth
self.cookiejar = cookiejar
self.sent = False
@@ -94,22 +95,20 @@ class Request(object):
_handlers = []
if self.auth or self.cookiejar:
if self.auth:
authr = urllib2.HTTPPasswordMgrWithDefaultRealm()
if self.auth:
if not isinstance(self.auth.handler, (urllib2.AbstractBasicAuthHandler, urllib2.AbstractDigestAuthHandler)):
auth_manager.add_password(self.auth.realm, self.url, self.auth.username, self.auth.password)
self.auth.handler = self.auth.handler(auth_manager)
auth_manager.add_auth(self.url, self.auth)
authr.add_password(None, self.url, self.auth[0], self.auth[1])
auth_handler = urllib2.HTTPBasicAuthHandler(authr)
_handlers.append(auth_handler)
if self.cookiejar:
cookie_handler = urllib2.HTTPCookieProcessor(cookiejar)
_handlers.append(cookie_handler)
_handlers.append(self.auth.handler)
if self.cookiejar:
cookie_handler = urllib2.HTTPCookieProcessor(cookiejar)
_handlers.append(cookie_handler)
if _handlers:
opener = urllib2.build_opener(*_handlers)
return opener.open
else:
return urllib2.urlopen
@@ -118,7 +117,7 @@ class Request(object):
"""Build internal Response object from given response."""
self.response.status_code = resp.code
self.response.headers = resp.info().dict
self.response.headers = resp.info().dict or resp.headers
self.response.content = resp.read()
self.response.url = resp.url
@@ -162,7 +161,8 @@ class Request(object):
else:
self._build_response(resp)
self.response.ok = True
self.response.cached = False
self.response.cached = False
else:
self.response.cached = True
@@ -199,6 +199,136 @@ class Response(object):
if self.error:
raise self.error
class AuthManager(object):
def __new__(cls):
singleton = cls.__dict__.get('__singleton__')
if singleton is not None:
return singleton
cls.__singleton__ = singleton = object.__new__(cls)
return singleton
def __init__(self):
self.passwd = {}
self._auth = {}
def add_auth(self, uri, auth):
uri = self.reduce_uri(uri, False)
self._auth[uri] = auth
def add_password(self, realm, uri, user, passwd):
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
reduced_uri = tuple([self.reduce_uri(u, False) for u in uri])
if reduced_uri not in self.passwd:
self.passwd[reduced_uri] = {}
self.passwd[reduced_uri] = (user, passwd)
def find_user_password(self, realm, authuri):
for uris, authinfo in self.passwd.iteritems():
reduced_authuri = self.reduce_uri(authuri, False)
for uri in uris:
if self.is_suburi(uri, reduced_authuri):
return authinfo
return (None, None)
def get_auth(self, uri):
uri = self.reduce_uri(uri, False)
return self._auth.get(uri, None)
def reduce_uri(self, uri, default_port=True):
"""Accept authority or URI and extract only the authority and path."""
# note HTTP URLs do not have a userinfo component
parts = urllib2.urlparse.urlsplit(uri)
if parts[1]:
# URI
scheme = parts[0]
authority = parts[1]
path = parts[2] or '/'
else:
# host or host:port
scheme = None
authority = uri
path = '/'
host, port = urllib2.splitport(authority)
if default_port and port is None and scheme is not None:
dport = {"http": 80,
"https": 443,
}.get(scheme)
if dport is not None:
authority = "%s:%d" % (host, dport)
return authority, path
def is_suburi(self, base, test):
"""Check if test is below base in a URI tree
Both args must be URIs in reduced form.
"""
if base == test:
return True
if base[0] != test[0]:
return False
common = urllib2.posixpath.commonprefix((base[1], test[1]))
if len(common) == len(base[1]):
return True
return False
def empty(self):
self.passwd = {}
def remove(self, uri, realm=None):
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
for default_port in True, False:
reduced_uri = tuple([self.reduce_uri(u, default_port) for u in uri])
del self.passwd[reduced_uri][realm]
def __contains__(self, uri):
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
uri = tuple([self.reduce_uri(u, False) for u in uri])
if uri in self.passwd:
return True
return False
auth_manager = AuthManager()
class AuthObject(object):
"""The :class:`AuthObject` is a simple HTTP Authentication token. When
given to a Requests function, it enables Basic HTTP Authentication for that
Request. You can also enable Authorization for domain realms with AutoAuth.
See AutoAuth for more details.
:param username: Username to authenticate with.
:param password: Password for given username.
:param realm: (optional) the realm this auth applies to
:param handler: (optional) basic || digest || proxy_basic || proxy_digest
"""
_handlers = {
'basic': urllib2.HTTPBasicAuthHandler,
'digest': urllib2.HTTPDigestAuthHandler,
'proxy_basic': urllib2.ProxyBasicAuthHandler,
'proxy_digest': urllib2.ProxyDigestAuthHandler
}
def __init__(self, username, password, handler='basic', realm=None):
self.username = username
self.password = password
self.realm = realm
self.handler = self._handlers.get(handler.lower(), urllib2.HTTPBasicAuthHandler)
def request(method, url, **kwargs):
"""Sends a `method` request. Returns :class:`Response` object.
@@ -216,7 +346,7 @@ def request(method, url, **kwargs):
r = Request(method=method, url=url, data=data, headers=kwargs.pop('headers', {}),
cookiejar=kwargs.pop('cookies', None), files=kwargs.pop('files', None),
auth=_detect_auth(url, kwargs.pop('auth', None)))
auth=kwargs.pop('auth', auth_manager.get_auth(url)))
r.send()
return r.response
@@ -231,8 +361,7 @@ def get(url, params={}, headers={}, cookies=None, auth=None):
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
"""
return request('GET', url, params=params, headers=headers, cookiejar=cookies,
auth=_detect_auth(url, auth))
return request('GET', url, params=params, headers=headers, cookiejar=cookies, auth=auth)
def head(url, params={}, headers={}, cookies=None, auth=None):
@@ -245,8 +374,7 @@ def head(url, params={}, headers={}, cookies=None, auth=None):
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
"""
return request('HEAD', url, params=params, headers=headers, cookiejar=cookies,
auth=_detect_auth(url, auth))
return request('HEAD', url, params=params, headers=headers, cookiejar=cookies, auth=auth)
def post(url, data={}, headers={}, files=None, cookies=None, auth=None):
@@ -260,8 +388,7 @@ def post(url, data={}, headers={}, files=None, cookies=None, auth=None):
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
"""
return request('POST', url, data=data, headers=headers, files=files, cookiejar=cookies,
auth=_detect_auth(url, auth))
return request('POST', url, data=data, headers=headers, files=files, cookiejar=cookies, auth=auth)
def put(url, data=b'', headers={}, files={}, cookies=None, auth=None):
@@ -275,8 +402,7 @@ def put(url, data=b'', headers={}, files={}, cookies=None, auth=None):
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
"""
return request('PUT', url, data=data, headers=headers, files=files, cookiejar=cookies,
auth=_detect_auth(url, auth))
return request('PUT', url, data=data, headers=headers, files=files, cookiejar=cookies, auth=auth)
def delete(url, params={}, headers={}, cookies=None, auth=None):
@@ -289,48 +415,7 @@ def delete(url, params={}, headers={}, cookies=None, auth=None):
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
"""
return request('DELETE', url, params=params, headers=headers, cookiejar=cookies,
auth=_detect_auth(url, auth))
def add_autoauth(url, authobject):
"""Registers given AuthObject to given URL domain. for auto-activation.
Once a URL is registered with an AuthObject, the configured HTTP
Authentication will be used for all requests with URLS containing the given
URL string.
Example: ::
>>> c_auth = requests.AuthObject('kennethreitz', 'xxxxxxx')
>>> requests.add_autoauth('https://convore.com/api/', c_auth)
>>> r = requests.get('https://convore.com/api/account/verify.json')
# Automatically HTTP Authenticated! Wh00t!
:param url: Base URL for given AuthObject to auto-activate for.
:param authobject: AuthObject to auto-activate.
"""
global AUTOAUTHS
AUTOAUTHS.append((url, authobject))
def _detect_auth(url, auth):
"""Returns registered AuthObject for given url if available, defaulting to
given AuthObject.
"""
return _get_autoauth(url) if not auth else auth
def _get_autoauth(url):
"""Returns registered AuthObject for given url if available."""
for (autoauth_url, auth) in AUTOAUTHS:
if autoauth_url in url:
return auth
return None
return request('DELETE', url, params=params, headers=headers, cookiejar=cookies, auth=auth)
class RequestException(Exception):
"""There was an ambiguous exception that occured while handling your
test_requests.py
+1 -3
View File
@@ -42,13 +42,11 @@ class RequestsTestSuite(unittest.TestCase):
self.assertEqual(r.status_code, 200)
requests.add_autoauth(url, auth)
r = requests.get(url)
self.assertEqual(r.status_code, 200)
# reset auto authentication
requests.AUTOAUTHS = []
requests.auth_manager.empty()
def test_POSTBIN_GET_POST_FILES(self):
bin = requests.get('http://www.postbin.org/')