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
127
128
129
130
131
132
133
134
135
136
|
import cgi
import json
import re
import urllib.parse
from datetime import datetime
from typing import Any, Callable, Dict, Tuple
from szurubooru import db
from szurubooru.func import util
from szurubooru.rest import context, errors, middleware, routes
def _json_serializer(obj: Any) -> str:
""" 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: Any) -> str:
return json.dumps(obj, default=_json_serializer, indent=2)
def _get_headers(env: Dict[str, Any]) -> Dict[str, str]:
headers = {} # type: Dict[str, str]
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: Dict[str, Any]) -> context.Context:
method = env["REQUEST_METHOD"]
path = "/" + env["PATH_INFO"].lstrip("/")
path = path.encode("latin-1").decode("utf-8") # PEP-3333
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(env, method, path, headers, params, files)
def application(
env: Dict[str, Any], start_response: Callable[[str, Any], Any]
) -> Tuple[bytes]:
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, ex_handler in errors.error_handlers.items():
if isinstance(ex, exception_type):
ex_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"),)
|
