__version__ = '0.1'
__all__ = [
'Resource', 'URL', 'Session', 'exceptions',
]
import crookbook
import requests
import requests.exceptions
import six
from six.moves.urllib import parse as urlparse
@crookbook.described('for {0.__url__}')
class Unit(object):
'''Base class where common Lamium objects share code - not intended
to be inherited directly outside of this module.'''
def __init__(self, session, url):
if not (isinstance(session, Session)):
err = 'session must be a Lamium session object, not %s' % type(session)
raise ValueError(err)
if not isinstance(url, six.string_types):
err = 'url must be a string type, not %s' % type(url)
self.__session__ = session
self.__url__ = url
def __call__(self, *args, **kwargs):
'''Merge additional parameters against the current object to create a
new instance of the same class combining the two. This will use
Session.format_url.'''
fmt = self.__session__.format_url
url = fmt(self.__url__, args, kwargs)
return self.at(url)
# Subclasses should use this to build another object of a similar
# type at the given URL. By default, it presumes it can construct
# an object of the same class as itself, passing the connection
# object and the URL path as positional arguments.
def at(self, url):
'''Constructs another instance of the same class which points
to the provided URL - it will be linked to the same underlying
session.'''
return self.__class__(self.__session__, url)
def __str__(self):
return self.__url__
[docs]@crookbook.essence('__session__ __url__', mutable=False)
class URL(Unit):
'''Lightweight object used for easy construction of URLs, which can then
be easily turned into a Resource object.'''
@property
def deURL(self):
return self.__session__.at(self.__url__)
def __getattr__(self, name):
if name in self.__session__.__resource_delegates__:
return getattr(self.deURL, name)
if name.startswith('__') and name.endswith('__'):
raise AttributeError("%r: %s" % (self, name))
return self(name)
_verb_func = '''
def {0}(self, *args, **kwargs):
"""Alias for session.{0} which passes the URL of this object as the
first argument."""
return self.request("{0}", *args, **kwargs)
'''
_verb_nobody_func = '''
def {0}(self, **kwargs):
"""XXX."""
return self.load_response(self.request("{1}", **kwargs))
'''
_verb_body_func = '''
def {0}(self, *args, **kwargs):
"""XXX."""
return self.send_request("{1}", *args, **kwargs)
'''
class LamiumResourceMeta(type):
def __init__(cls, name, bases, nmspec):
super(LamiumResourceMeta, cls).__init__(name, bases, nmspec)
compiled = {}
for verb in cls.__verbs__:
if hasattr(cls, verb):
continue
fcode = compile(_verb_func.format(verb), '<lamium_dynfunc>', 'exec')
compiled[verb] = fcode
for verb in cls.__verbs__:
mverb = verb.lower()
if hasattr(cls, mverb):
continue
functmpl = _verb_body_func if verb in cls.__verbs_with_bodies__ else _verb_body_func
fcode = compile(functmpl.format(mverb, verb), '<lamium_dynfunc>', 'exec')
compiled[mverb] = fcode
ns = {}
for method_name, fcode in compiled.items():
eval(fcode, {}, ns)
setattr(cls, method_name, ns[method_name])
#del _verb_func, _verb_body_func, _verb_nobody_func
class BaseResource(six.with_metaclass(LamiumResourceMeta, Unit)):
# How do you do frozensets these days?
__verbs__ = frozenset('DELETE GET HEAD PATCH POST PUT'.split())
__verbs_with_bodies__ = frozenset('PATCH POST PUT'.split())
@property
def URL(self):
return URL(self.__session__, self.__url__)
def request(self, method, *args, **kwargs):
return self.__session__.request(method, self.__url__, *args, **kwargs)
def at(self, url):
return self.__session__.at(url)
# send_request will be defined in superclass, won't have "response" or
# notfound behaviour. That'll be defined by us.
def send_request(self, method, *args, **kwargs):
kwargs = self._merge_request_params(*args, **kwargs)
return self.load_response(self.request(method, **kwargs))
def load_response(self, response):
return response
def _merge_request_params(self, data=None, **kwargs):
if data is not None:
kwargs['data'] = data
return kwargs
_NOTGIVEN = object()
[docs]class Resource(BaseResource):
[docs] def get(self, *args, **kwargs):
# Emulating Tortilla here.
if args:
return self(*args).get(**kwargs) #pylint: disable=no-member
return super(Resource, self).get(**kwargs) #pylint: disable=no-member
[docs] def send_request(self, method, data=None, notfound=None, response=False, **kwargs):
params = self._merge_request_params(data, **kwargs)
resp = self.request(method, **params)
if response:
return resp
try:
return self.load_response(resp)
except (self.exceptions.NotFound, self.exceptions.Gone):
if notfound is not _NOTGIVEN:
return notfound
raise
[docs] def load_response(self, response):
if 400 <= response.status_code < 600:
self.raise_for_status(response)
return response
[docs] def raise_for_status(self, response):
raise self.exceptions.exception_for_code(response.status_code)(response)
def _merge_request_params(self, data=None, **kwargs):
if data is None:
if 'json' in kwargs:
raise ValueError('cannot define json as positional and keyword argument')
return {'json': kwargs}
return kwargs
[docs]class Session(object):
resource_class = BaseResource
def __init__(self, req_sess=None, resource_class=None):
if resource_class is not None:
self.resource_class = resource_class
self.req_sess = req_sess or requests.Session()
self.__resource_delegates__ = frozenset(
list(self.resource_class.__verbs__) +
[x.lower() for x in self.resource_class.__verbs__] +
['request']
)
[docs] def request(self, method, url, *args, **kwargs):
return self.req_sess.request(method, url, *args, **kwargs)
[docs] def at(self, url):
return self.resource_class(self, url)
[docs] @classmethod
def Root(cls, url, **kwargs):
return cls(**kwargs).at(url)
# In a Python 3 only world, this would preferably be types.SimpleNamespace.
[docs]class exceptions(object):
'''Namespace containing exception classes used by lamium module.'''
[docs] class ErrorResponse(requests.exceptions.HTTPError):
def __init__(self, response):
self.status_code = response.status_code
self.reason = response.raw.reason
err = '{0.status_code} - {0.reason}'.format(self)
requests.exceptions.HTTPError.__init__(self, err, response=response)
[docs] class ClientError(ErrorResponse): pass
[docs] class BadRequest(ClientError): pass
[docs] class Unauthorized(ClientError): pass
[docs] class Forbidden(ClientError): pass
[docs] class NotFound(ClientError): pass
[docs] class MethodNotAllowed(ClientError): pass
[docs] class Conflict(ClientError): pass
[docs] class Gone(ClientError): pass
[docs] class ServerError(ErrorResponse): pass
Timeout = requests.exceptions.Timeout
[docs] @classmethod
def exception_for_code(cls, code):
# Do we have a specific class for this code?
exc_class = {
400: cls.BadRequest,
401: cls.Unauthorized,
403: cls.Forbidden,
404: cls.NotFound,
405: cls.MethodNotAllowed,
409: cls.Conflict,
410: cls.Gone,
}.get(code, None)
# If not, try something a bit more generic.
if exc_class is None:
exc_class = {
4: cls.ClientError,
5: cls.ServerError,
}.get(code // 100, None)
return exc_class or cls.ErrorResponse
Session.exceptions = Resource.exceptions = exceptions
#
# Exception handling.
#