Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def parse_json(self, req, name, field):
"""Pull a json value from the request."""
json_data = self._cache.get("json")
if json_data is None:
try:
self._cache["json"] = json_data = parse_json_body(req)
except json.JSONDecodeError as e:
return self.handle_invalid_json_error(e, req)
if json_data is None:
return core.missing
return core.get_value(json_data, name, field, allow_many_nested=True)
From Flask-Restful. See NOTICE file for license information.
"""
try:
flask.abort(http_status_code)
except HTTPException as err:
err.data = kwargs
err.exc = exc
raise err
def is_json_request(req):
return core.is_json(req.mimetype)
class FlaskParser(core.Parser):
"""Flask request argument parser."""
__location_map__ = dict(
view_args="parse_view_args",
path="parse_view_args",
**core.Parser.__location_map__
)
def parse_view_args(self, req, name, field):
"""Pull a value from the request's ``view_args``."""
return core.get_value(req.view_args, name, field)
def parse_json(self, req, name, field):
"""Pull a json value from the request."""
json_data = self._cache.get("json")
if json_data is None:
def parse_form(self, req, name, field):
"""
Intended to emulate parsing SNCL arguments from FDSNWS formatted
postfiles.
:param req: Request object to be parsed
:type req: :py:class:`flask.Request`
See also:
http://www.fdsn.org/webservices/FDSN-WS-Specifications-1.1.pdf
"""
return webargs.core.get_value(
self._parse_postfile(self._get_data(req)), name, field)
from aiohttp import web
from webargs import core
from webargs.async import AsyncParser
class HTTPUnprocessableEntity(web.HTTPClientError):
status_code = 422
class AIOHTTPParser(AsyncParser):
"""aiohttp request argument parser."""
__location_map__ = dict(
match_info='parse_match_info',
**core.Parser.__location_map__
)
def parse_querystring(self, req, name, field):
"""Pull a querystring value from the request."""
return core.get_value(req.GET, name, field)
@asyncio.coroutine
def parse_form(self, req, name, field):
"""Pull a form value from the request."""
post_data = self._cache.get('post')
if post_data is None:
pass
self._cache['post'] = req.POST
return core.get_value(self._cache['post'], name, field)
@asyncio.coroutine
def parse_json(self, req, name, field):
"""Pull a json value from the request."""
json_data = self._cache.get("json")
if json_data is None:
try:
self._cache["json"] = json_data = core.parse_json(req.body)
except json.JSONDecodeError as e:
if e.doc == "":
return core.missing
else:
raise
return core.get_value(json_data, name, field, allow_many_nested=True)
def annotations2schema(
func: typing.Callable, type_mapping: TypeMapping = None
) -> Schema:
type_mapping = type_mapping or DEFAULT_TYPE_MAPPING
annotations = getattr(func, "__annotations__", {})
signature = inspect.signature(func)
fields_dict = {}
for name, annotation in annotations.items():
# Skip over request argument and return annotation
if name == "return" or (
isinstance(annotation, type) and issubclass(annotation, Request)
):
continue
fields_dict[name] = _type2field(name, annotation, signature, type_mapping)
return core.dict2schema(fields_dict)
def parse_json_body(req):
if req.content_length in (None, 0):
# Nothing to do
return {}
if is_json_request(req):
body = req.stream.read()
if body:
try:
return core.parse_json(body)
except json.JSONDecodeError as e:
if e.doc == "":
return core.missing
else:
raise
return {}
def parse_json(self, req, name, field):
"""Pull a json value from the request body."""
json_data = self._cache.get("json")
if json_data is None:
try:
self._cache["json"] = json_data = core.parse_json(req.body)
except AttributeError:
return core.missing
except json.JSONDecodeError as e:
if e.doc == "":
return core.missing
else:
return self.handle_invalid_json_error(e, req)
return core.get_value(json_data, name, field, allow_many_nested=True)
def parse_cookies(self, req: Request, name: str, field: Field) -> typing.Any:
"""Pull a value from the cookiejar."""
return core.get_value(req.cookies, name, field)
def parse_querystring(self, req, name, field):
"""Pull a querystring value from the request."""
return core.get_value(req.args, name, field)