Source code for webargs.async_decorators

import inspect
import collections
import functools

from webargs import core


def _use_args(self, argmap, req=None, locations=None, as_kwargs=False, validate=None):
    """Decorator that injects parsed arguments into a view function or method.

    Receives the same arguments as `webargs.core.Parser.use_args`.
    """
    locations = locations or self.locations
    request_obj = req
    # Optimization: If argmap is passed as a dictionary, we only need
    # to generate a Schema once
    if isinstance(argmap, collections.Mapping):
        argmap = core.argmap2schema(argmap)()

    def decorator(func):
        req_ = request_obj

        if inspect.iscoroutinefunction(func):
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                req_obj = req_

                # if as_kwargs is passed, must include all args
                force_all = as_kwargs

                if not req_obj:
                    req_obj = self.get_request_from_view_args(func, args, kwargs)
                # NOTE: At this point, argmap may be a Schema, callable, or dict
                parsed_args = await self.parse(argmap,
                                                    req=req_obj, locations=locations,
                                                    validate=validate, force_all=force_all)
                if as_kwargs:
                    kwargs.update(parsed_args)
                    return await func(*args, **kwargs)
                else:
                    # Add parsed_args after other positional arguments
                    new_args = args + (parsed_args, )
                    return await func(*new_args, **kwargs)
        else:
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                req_obj = req_

                # if as_kwargs is passed, must include all args
                force_all = as_kwargs

                if not req_obj:
                    req_obj = self.get_request_from_view_args(func, args, kwargs)
                # NOTE: At this point, argmap may be a Schema, callable, or dict
                parsed_args = yield from self.parse(argmap,
                                                    req=req_obj, locations=locations,
                                                    validate=validate, force_all=force_all)
                if as_kwargs:
                    kwargs.update(parsed_args)
                    return func(*args, **kwargs)
                else:
                    # Add parsed_args after other positional arguments
                    new_args = args + (parsed_args, )
                    return func(*new_args, **kwargs)
        wrapper.__wrapped__ = func
        return wrapper
    return decorator