Source code for wsme.api

import traceback
import functools
import inspect
import logging

import wsme.exc
import wsme.types

from wsme import utils

log = logging.getLogger(__name__)


def iswsmefunction(f):
    return hasattr(f, '_wsme_definition')


def wrapfunc(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        return f(*args, **kwargs)
    wrapper._wsme_original_func = f
    return wrapper


def getargspec(f):
    f = getattr(f, '_wsme_original_func', f)
    return inspect.getargspec(f)


[docs]class FunctionArgument(object): """ An argument definition of an api entry """ def __init__(self, name, datatype, mandatory, default): #: argument name self.name = name #: Data type self.datatype = datatype #: True if the argument is mandatory self.mandatory = mandatory #: Default value if argument is omitted self.default = default def resolve_type(self, registry): self.datatype = registry.resolve_type(self.datatype)
[docs]class FunctionDefinition(object): """ An api entry definition """ def __init__(self, func): #: Function name self.name = func.__name__ #: Function documentation self.doc = func.__doc__ #: Return type self.return_type = None #: The function arguments (list of :class:`FunctionArgument`) self.arguments = [] #: If the body carry the datas of a single argument, its type self.body_type = None #: Status code self.status_code = 200 #: True if extra arguments should be ignored, NOT inserted in #: the kwargs of the function and not raise UnknownArgument #: exceptions self.ignore_extra_args = False #: name of the function argument to pass the host request object. #: Should be set by using the :class:`wsme.types.HostRequest` type #: in the function @\ :function:`signature` self.pass_request = False #: Dictionnary of protocol-specific options. self.extra_options = None @staticmethod
[docs] def get(func): """ Returns the :class:`FunctionDefinition` of a method. """ if not hasattr(func, '_wsme_definition'): fd = FunctionDefinition(func) func._wsme_definition = fd return func._wsme_definition
[docs] def get_arg(self, name): """ Returns a :class:`FunctionArgument` from its name """ for arg in self.arguments: if arg.name == name: return arg return None
def resolve_types(self, registry): self.return_type = registry.resolve_type(self.return_type) self.body_type = registry.resolve_type(self.body_type) for arg in self.arguments: arg.resolve_type(registry) def set_options(self, body=None, ignore_extra_args=False, status_code=200, rest_content_types=('json', 'xml'), **extra_options): self.body_type = body self.status_code = status_code self.ignore_extra_args = ignore_extra_args self.rest_content_types = rest_content_types self.extra_options = extra_options def set_arg_types(self, argspec, arg_types): args, varargs, keywords, defaults = argspec if args[0] == 'self': args = args[1:] arg_types = list(arg_types) if self.body_type is not None: arg_types.append(self.body_type) for i, argname in enumerate(args): datatype = arg_types[i] mandatory = defaults is None or i < (len(args) - len(defaults)) default = None if not mandatory: default = defaults[i - (len(args) - len(defaults))] if datatype is wsme.types.HostRequest: self.pass_request = argname else: self.arguments.append(FunctionArgument(argname, datatype, mandatory, default))
class signature(object): """ Decorator that specify the argument types of an exposed function. :param return_type: Type of the value returned by the function :param argN: Type of the Nth argument :param body: If the function takes a final argument that is supposed to be the request body by itself, its type. :param status: HTTP return status code of the function. :param ignore_extra_args: Allow extra/unknow arguments (default to False) Most of the time this decorator is not supposed to be used directly, unless you are not using WSME on top of another framework. If an adapter is used, it will provide either a specialised version of this decororator, either a new decorator named @wsexpose that takes the same parameters (it will in addition expose the function, hence its name). """ def __init__(self, *types, **options): self.return_type = types[0] if types else None self.arg_types = [] if len(types) > 1: self.arg_types.extend(types[1:]) if 'body' in options: self.arg_types.append(options['body']) self.wrap = options.pop('wrap', False) self.options = options def __call__(self, func): argspec = getargspec(func) if self.wrap: func = wrapfunc(func) fd = FunctionDefinition.get(func) if fd.extra_options is not None: raise ValueError("This function is already exposed") fd.return_type = self.return_type fd.set_options(**self.options) if self.arg_types: fd.set_arg_types(argspec, self.arg_types) return func sig = signature class Response(object): """ Object to hold the "response" from a view function """ def __init__(self, obj, status_code=None, error=None): #: Store the result object from the view self.obj = obj #: Store an optional status_code self.status_code = status_code #: Return error details #: Must be a dictionnary with the following keys: faultcode, #: faultstring and an optional debuginfo self.error = error def format_exception(excinfo, debug=False): """Extract informations that can be sent to the client.""" error = excinfo[1] code = getattr(error, 'code', None) if code and utils.is_valid_code(code) and utils.is_client_error(code): faultstring = error.faultstring if hasattr(error, 'faultstring') \ else str(error) r = dict(faultcode="Client", faultstring=faultstring) log.warning("Client-side error: %s" % r['faultstring']) r['debuginfo'] = None return r else: faultstring = str(error) debuginfo = "\n".join(traceback.format_exception(*excinfo)) log.error('Server-side error: "%s". Detail: \n%s' % ( faultstring, debuginfo)) r = dict(faultcode="Server", faultstring=faultstring) if debug: r['debuginfo'] = debuginfo else: r['debuginfo'] = None return r