-
-
Notifications
You must be signed in to change notification settings - Fork 1.5k
/
exceptions.py
271 lines (183 loc) · 6.08 KB
/
exceptions.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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
from asyncio import CancelledError
from typing import Any, Dict, Optional, Union
from sanic.helpers import STATUS_CODES
class RequestCancelled(CancelledError):
quiet = True
class ServerKilled(Exception):
...
class SanicException(Exception):
message: str = ""
def __init__(
self,
message: Optional[Union[str, bytes]] = None,
status_code: Optional[int] = None,
quiet: Optional[bool] = None,
context: Optional[Dict[str, Any]] = None,
extra: Optional[Dict[str, Any]] = None,
) -> None:
self.context = context
self.extra = extra
if message is None:
if self.message:
message = self.message
elif status_code is not None:
msg: bytes = STATUS_CODES.get(status_code, b"")
message = msg.decode("utf8")
super().__init__(message)
if status_code is not None:
self.status_code = status_code
# quiet=None/False/True with None meaning choose by status
if quiet or quiet is None and status_code not in (None, 500):
self.quiet = True
class NotFound(SanicException):
"""
**Status**: 404 Not Found
"""
status_code = 404
quiet = True
class BadRequest(SanicException):
"""
**Status**: 400 Bad Request
"""
status_code = 400
quiet = True
InvalidUsage = BadRequest
class BadURL(BadRequest):
...
class MethodNotAllowed(SanicException):
"""
**Status**: 405 Method Not Allowed
"""
status_code = 405
quiet = True
def __init__(self, message, method, allowed_methods):
super().__init__(message)
self.headers = {"Allow": ", ".join(allowed_methods)}
MethodNotSupported = MethodNotAllowed
class ServerError(SanicException):
"""
**Status**: 500 Internal Server Error
"""
status_code = 500
class ServiceUnavailable(SanicException):
"""
**Status**: 503 Service Unavailable
The server is currently unavailable (because it is overloaded or
down for maintenance). Generally, this is a temporary state.
"""
status_code = 503
quiet = True
class URLBuildError(ServerError):
"""
**Status**: 500 Internal Server Error
"""
status_code = 500
class FileNotFound(NotFound):
"""
**Status**: 404 Not Found
"""
def __init__(self, message, path, relative_url):
super().__init__(message)
self.path = path
self.relative_url = relative_url
class RequestTimeout(SanicException):
"""The Web server (running the Web site) thinks that there has been too
long an interval of time between 1) the establishment of an IP
connection (socket) between the client and the server and
2) the receipt of any data on that socket, so the server has dropped
the connection. The socket connection has actually been lost - the Web
server has 'timed out' on that particular socket connection.
"""
status_code = 408
quiet = True
class PayloadTooLarge(SanicException):
"""
**Status**: 413 Payload Too Large
"""
status_code = 413
quiet = True
class HeaderNotFound(BadRequest):
"""
**Status**: 400 Bad Request
"""
class InvalidHeader(BadRequest):
"""
**Status**: 400 Bad Request
"""
class RangeNotSatisfiable(SanicException):
"""
**Status**: 416 Range Not Satisfiable
"""
status_code = 416
quiet = True
def __init__(self, message, content_range):
super().__init__(message)
self.headers = {"Content-Range": f"bytes */{content_range.total}"}
ContentRangeError = RangeNotSatisfiable
class ExpectationFailed(SanicException):
"""
**Status**: 417 Expectation Failed
"""
status_code = 417
quiet = True
HeaderExpectationFailed = ExpectationFailed
class Forbidden(SanicException):
"""
**Status**: 403 Forbidden
"""
status_code = 403
quiet = True
class InvalidRangeType(RangeNotSatisfiable):
"""
**Status**: 416 Range Not Satisfiable
"""
status_code = 416
quiet = True
class PyFileError(Exception):
def __init__(self, file):
super().__init__("could not execute config file %s", file)
class Unauthorized(SanicException):
"""
**Status**: 401 Unauthorized
:param message: Message describing the exception.
:param status_code: HTTP Status code.
:param scheme: Name of the authentication scheme to be used.
When present, kwargs is used to complete the WWW-Authentication header.
Examples::
# With a Basic auth-scheme, realm MUST be present:
raise Unauthorized("Auth required.",
scheme="Basic",
realm="Restricted Area")
# With a Digest auth-scheme, things are a bit more complicated:
raise Unauthorized("Auth required.",
scheme="Digest",
realm="Restricted Area",
qop="auth, auth-int",
algorithm="MD5",
nonce="abcdef",
opaque="zyxwvu")
# With a Bearer auth-scheme, realm is optional so you can write:
raise Unauthorized("Auth required.", scheme="Bearer")
# or, if you want to specify the realm:
raise Unauthorized("Auth required.",
scheme="Bearer",
realm="Restricted Area")
"""
status_code = 401
quiet = True
def __init__(self, message, status_code=None, scheme=None, **kwargs):
super().__init__(message, status_code)
# if auth-scheme is specified, set "WWW-Authenticate" header
if scheme is not None:
values = ['{!s}="{!s}"'.format(k, v) for k, v in kwargs.items()]
challenge = ", ".join(values)
self.headers = {
"WWW-Authenticate": f"{scheme} {challenge}".rstrip()
}
class LoadFileException(SanicException):
pass
class InvalidSignal(SanicException):
pass
class WebsocketClosed(SanicException):
quiet = True
message = "Client has closed the websocket connection"