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
114
115
116
117
118
119
120
121
122
123
124
125
126
|
import urllib.parse
import cgi
import json
import re
from datetime import datetime
from szurubooru import db
from szurubooru.func import util
from szurubooru.rest import errors, middleware, routes, context
def _json_serializer(obj):
''' JSON serializer for objects not serializable by default JSON code '''
if isinstance(obj, datetime):
serial = obj.isoformat('T') + 'Z'
return serial
raise TypeError('Type not serializable')
def _dump_json(obj):
return json.dumps(obj, default=_json_serializer, indent=2)
def _get_headers(env):
headers = {}
for key, value in env.items():
if key.startswith('HTTP_'):
key = util.snake_case_to_upper_train_case(key[5:])
headers[key] = value
return headers
def _create_context(env):
method = env['REQUEST_METHOD']
path = '/' + env['PATH_INFO'].lstrip('/')
headers = _get_headers(env)
files = {}
params = dict(urllib.parse.parse_qsl(env.get('QUERY_STRING', '')))
if 'multipart' in env.get('CONTENT_TYPE', ''):
form = cgi.FieldStorage(fp=env['wsgi.input'], environ=env)
if not form.list:
raise errors.HttpBadRequest(
'ValidationError', 'No files attached.')
body = form.getvalue('metadata')
for key in form:
files[key] = form.getvalue(key)
else:
body = env['wsgi.input'].read()
if body:
try:
if isinstance(body, bytes):
body = body.decode('utf-8')
for key, value in json.loads(body).items():
params[key] = value
except (ValueError, UnicodeDecodeError):
raise errors.HttpBadRequest(
'ValidationError',
'Could not decode the request body. The JSON '
'was incorrect or was not encoded as UTF-8.')
return context.Context(method, path, headers, params, files)
def application(env, start_response):
try:
ctx = _create_context(env)
if 'application/json' not in ctx.get_header('Accept'):
raise errors.HttpNotAcceptable(
'ValidationError',
'This API only supports JSON responses.')
for url, allowed_methods in routes.routes.items():
match = re.fullmatch(url, ctx.url)
if match:
if ctx.method not in allowed_methods:
raise errors.HttpMethodNotAllowed(
'ValidationError',
'Allowed methods: %r' % allowed_methods)
handler = allowed_methods[ctx.method]
break
else:
raise errors.HttpNotFound(
'ValidationError',
'Requested path ' + ctx.url + ' was not found.')
try:
ctx.session = db.session()
try:
for hook in middleware.pre_hooks:
hook(ctx)
try:
response = handler(ctx, match.groupdict())
except Exception:
ctx.session.rollback()
raise
finally:
for hook in middleware.post_hooks:
hook(ctx)
finally:
db.session.remove()
start_response('200', [('content-type', 'application/json')])
return (_dump_json(response).encode('utf-8'),)
except Exception as ex:
for exception_type, handler in errors.error_handlers.items():
if isinstance(ex, exception_type):
handler(ex)
raise
except errors.BaseHttpError as ex:
start_response(
'%d %s' % (ex.code, ex.reason),
[('content-type', 'application/json')])
blob = {
'name': ex.name,
'title': ex.title,
'description': ex.description,
}
if ex.extra_fields is not None:
for key, value in ex.extra_fields.items():
blob[key] = value
return (_dump_json(blob).encode('utf-8'),)
|
