-
-
Notifications
You must be signed in to change notification settings - Fork 155
/
flaskparser.py
113 lines (84 loc) · 3.28 KB
/
flaskparser.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
# -*- coding: utf-8 -*-
"""Flask request argument parsing module.
Example: ::
from flask import Flask
from webargs import fields
from webargs.flaskparser import use_args
app = Flask(__name__)
hello_args = {
'name': fields.Str(required=True)
}
@app.route('/')
@use_args(hello_args)
def index(args):
return 'Hello ' + args['name']
"""
import flask
from werkzeug.exceptions import HTTPException
from webargs import core
from webargs.multidictproxy import MultiDictProxy
def abort(http_status_code, exc=None, **kwargs):
"""Raise a HTTPException for the given http_status_code. Attach any keyword
arguments to the exception for later processing.
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="load_view_args",
path="load_view_args",
**core.Parser.__location_map__
)
def _raw_load_json(self, req):
"""Return a json payload from the request for the core parser's load_json
Checks the input mimetype and may return 'missing' if the mimetype is
non-json, even if the request body is parseable as json."""
if not is_json_request(req):
return core.missing
return core.parse_json(req.get_data(cache=True))
def _handle_invalid_json_error(self, error, req, *args, **kwargs):
abort(400, exc=error, messages={"json": ["Invalid JSON body."]})
def load_view_args(self, req, schema):
"""Return the request's ``view_args`` or ``missing`` if there are none."""
return req.view_args or core.missing
def load_querystring(self, req, schema):
"""Return query params from the request as a MultiDictProxy."""
return MultiDictProxy(req.args, schema)
def load_form(self, req, schema):
"""Return form values from the request as a MultiDictProxy."""
return MultiDictProxy(req.form, schema)
def load_headers(self, req, schema):
"""Return headers from the request as a MultiDictProxy."""
return MultiDictProxy(req.headers, schema)
def load_cookies(self, req, schema):
"""Return cookies from the request."""
return req.cookies
def load_files(self, req, schema):
"""Return files from the request as a MultiDictProxy."""
return MultiDictProxy(req.files, schema)
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 422 error.
"""
status_code = error_status_code or self.DEFAULT_VALIDATION_STATUS
abort(
status_code,
exc=error,
messages=error.messages,
schema=schema,
headers=error_headers,
)
def get_default_request(self):
"""Override to use Flask's thread-local request objec by default"""
return flask.request
parser = FlaskParser()
use_args = parser.use_args
use_kwargs = parser.use_kwargs