# -*- coding: utf-8 -*-
"""Pyramid request argument parsing.
Example usage: ::
from wsgiref.simple_server import make_server
from pyramid.config import Configurator
from pyramid.response import Response
from marshmallow import fields
from webargs.pyramidparser import use_args
hello_args = {
'name': fields.Str(missing='World')
}
@use_args(hello_args)
def hello_world(request, args):
return Response('Hello ' + args['name'])
if __name__ == '__main__':
config = Configurator()
config.add_route('hello', '/')
config.add_view(hello_world, route_name='hello')
app = config.make_wsgi_app()
server = make_server('0.0.0.0', 6543, app)
server.serve_forever()
"""
import collections
import functools
from webob.multidict import MultiDict
from pyramid.httpexceptions import exception_response
from marshmallow.compat import text_type
from webargs import core
[docs]class PyramidParser(core.Parser):
"""Pyramid request argument parser."""
__location_map__ = dict(matchdict="parse_matchdict", **core.Parser.__location_map__)
[docs] def parse_querystring(self, req, name, field):
"""Pull a querystring value from the request."""
return core.get_value(req.GET, name, field)
[docs] def parse_json(self, req, name, field):
"""Pull a json value from the request."""
try:
json_data = req.json_body
except ValueError:
return core.missing
return core.get_value(json_data, name, field, allow_many_nested=True)
[docs] def parse_cookies(self, req, name, field):
"""Pull the value from the cookiejar."""
return core.get_value(req.cookies, name, field)
[docs] def parse_files(self, req, name, field):
"""Pull a file from the request."""
files = ((k, v) for k, v in req.POST.items() if hasattr(v, "file"))
return core.get_value(MultiDict(files), name, field)
[docs] def parse_matchdict(self, req, name, field):
"""Pull a value from the request's `matchdict`."""
return core.get_value(req.matchdict, name, field)
[docs] def handle_error(self, error, req, schema, error_status_code, error_headers):
"""Handles errors during parsing. Aborts the current HTTP request and
responds with a 400 error.
"""
status_code = error_status_code or getattr(error, "status_code", 422)
raise exception_response(
status_code, detail=text_type(error), headers=error_headers
)
[docs] def use_args(
self,
argmap,
req=None,
locations=core.Parser.DEFAULT_LOCATIONS,
as_kwargs=False,
validate=None,
force_all=None,
error_status_code=None,
error_headers=None,
):
"""Decorator that injects parsed arguments into a view callable.
Supports the *Class-based View* pattern where `request` is saved as an instance
attribute on a view class.
:param dict argmap: Either a `marshmallow.Schema`, a `dict`
of argname -> `marshmallow.fields.Field` pairs, or a callable
which accepts a request and returns a `marshmallow.Schema`.
:param req: The request object to parse. Pulled off of the view by default.
:param tuple locations: Where on the request to search for values.
:param bool as_kwargs: Whether to insert arguments as keyword arguments.
:param callable validate: Validation function that receives the dictionary
of parsed arguments. If the function returns ``False``, the parser
will raise a :exc:`ValidationError`.
:param bool force_all: If `True`, missing arguments will be included
in the parsed arguments dictionary with the ``missing`` value.
If `False`, missing values will be omitted. If `None`, fall back
to the value of ``as_kwargs``.
:param int error_status_code: Status code passed to error handler functions when
a `ValidationError` is raised.
:param dict error_headers: Headers passed to error handler functions when a
a `ValidationError` is raised.
"""
locations = locations or self.locations
# Optimization: If argmap is passed as a dictionary, we only need
# to generate a Schema once
if isinstance(argmap, collections.Mapping):
argmap = core.dict2schema(argmap)()
def decorator(func):
force_all_ = force_all if force_all is not None else as_kwargs
@functools.wraps(func)
def wrapper(obj, *args, **kwargs):
# The first argument is either `self` or `request`
try: # get self.request
request = req or obj.request
except AttributeError: # first arg is request
request = obj
# NOTE: At this point, argmap may be a Schema, callable, or dict
parsed_args = self.parse(
argmap,
req=request,
locations=locations,
validate=validate,
force_all=force_all_,
error_status_code=error_status_code,
error_headers=error_headers,
)
if as_kwargs:
kwargs.update(parsed_args)
return func(obj, *args, **kwargs)
else:
return func(obj, parsed_args, *args, **kwargs)
wrapper.__wrapped__ = func
return wrapper
return decorator
parser = PyramidParser()
use_args = parser.use_args
use_kwargs = parser.use_kwargs