6 _Request = collections.namedtuple(
27 class Request(_Request):
28 def __new__(cls, method, path, env=None):
34 accept = env.get('HTTP_ACCEPT')
35 accept_encoding = env.get('HTTP_ACCEPT_ENCODING')
36 accept_language = env.get('HTTP_ACCEPT_LANGUAGE')
37 content = env.get('CONTENT', '')
38 content_type = env.get('CONTENT_TYPE')
39 query = env.get('QUERY_STRING')
40 user_agent = env.get('HTTP_USER_AGENT')
42 content_length = env.get('CONTENT_LENGTH')
44 if content_length == '' or content_length is None:
48 content_length = int(content_length)
50 errors.append('Unable to parse Content-Length "{}"'.format(content_length))
54 cookie = http.cookies.SimpleCookie(env.get('HTTP_COOKIE'))
56 cookie = http.cookies.SimpleCookie()
59 GET = urllib.parse.parse_qs(query)
62 errors.append('Unable to parse GET parameters from query string "{}"'.format(query))
66 if content_type == 'application/x-www-form-urlencoded':
67 POST = urllib.parse.parse_qs(content)
70 errors.append('Unable to parse POST parameters from content string "{}"'.format(content))
74 errors.append('Unable to parse POST parameters from content string "{}"'.format(content))
81 elif method == 'POST':
86 result = super().__new__(
92 accept_encoding=accept_encoding,
93 accept_language=accept_language,
95 content_length = content_length,
96 content_type = content_type,
99 parameters=parameters,
102 user_agent=user_agent,
105 if path.startswith('/'):
106 result.subpath = path[1:]
108 result.subpath = path
112 def _get_request_from_env(env):
113 method = env.get('REQUEST_METHOD')
114 path = env.get('PATH_INFO')
115 return Request(method, path, env)
117 _Response = collections.namedtuple(
127 class Response(_Response):
128 def __new__(cls, content, **kwargs):
129 status = kwargs.pop('status', 200)
130 assert isinstance(status, int)
132 content_type = kwargs.pop('content_type')
133 assert isinstance(content_type, str)
135 extra_headers = kwargs.pop('extra_headers', {})
136 assert isinstance(extra_headers, dict)
138 assert len(kwargs) == 0
140 return super().__new__(
143 content_type=content_type,
144 extra_headers=extra_headers,
150 # Start with the defaults
152 'X-Content-Type-Options': 'nosniff',
155 result = {**result, **(self.extra_headers)}
158 'Content-Type': self.content_type,
161 for key, value in builtin_headers.items():
163 raise Exception('Header "{}" defined twice'.format(key))
167 return tuple(sorted(result.items()))
170 class HTMLResponse(Response):
171 def __new__(cls, content, **kwargs):
172 assert 'content_type' not in kwargs
174 return super().__new__(
177 content_type='text/html; charset=utf-8',
181 class JSONResponse(Response):
182 def __new__(cls, content_json, **kwargs):
183 assert 'content_type' not in kwargs
184 assert 'content' not in kwargs
186 self = super().__new__(
188 content=json.dumps(content_json),
189 content_type='application/json; charset=utf-8',
192 self.content_json = content_json
195 class TextResponse(Response):
196 def __new__(cls, content, **kwargs):
197 assert 'content_type' not in kwargs
199 return super().__new__(
202 content_type='text/plain; charset=utf-8',
206 _RedirectResponse = collections.namedtuple(
214 class RedirectResponse(_RedirectResponse):
215 def __new__(cls, location, **kwargs):
216 assert isinstance(location, str)
218 permanent = kwargs.pop('permanent', True)
219 assert isinstance(permanent, bool)
220 assert len(kwargs) == 0
222 return super().__new__(
230 return 308 if self.permanent else 307
234 return (('Location', self.location),)
240 def default_file_not_found_handler(request):
242 'Path "{}" with query "{}" not found'.format(request.path, request.query),
246 def route_on_subpath(**kwargs):
247 routes = kwargs.pop('routes')
248 file_not_found_handler = kwargs.pop(
249 'file_not_found_handler',
250 default_file_not_found_handler,
254 raise Exception('Keyword argument "routes" is required')
257 raise Exception('Unexpected keyword argument')
259 def wrapped(request):
260 split_subpath = request.subpath.split('/', 1)
261 subpath = split_subpath[0]
263 if len(split_subpath) == 2:
264 request.subpath = split_subpath[1]
268 return routes.get(subpath, file_not_found_handler)(request)
284 def default_method_not_allowed_handler(request):
285 return TextResponse('', status=405)
287 def default_options_handler(handlers):
288 def handler(request):
289 return Response(','.join(handlers.keys()))
292 def route_on_method(**kwargs):
294 for method in REQUEST_METHODS:
296 handlers[method] = kwargs.pop(method)
298 method_not_allowed_handler = kwargs.pop(
299 'method_not_allowed',
300 default_method_not_allowed_handler,
303 assert len(kwargs) == 0
305 if 'OPTIONS' not in handlers:
306 handlers['OPTIONS'] = default_options_handler(handlers)
308 def handler(request):
310 request.method.upper(),
311 method_not_allowed_handler,
316 def _get_status(response):
319 101: '101 Switching Protocols',
323 203: '203 Non-Authoritative Information',
324 204: '204 No Content',
325 205: '205 Reset Content',
326 300: '300 Multiple Choices',
327 301: '301 Moved Permanently',
328 304: '304 Not Modified',
329 307: '307 Temporary Redirect',
330 308: '308 Permanent Redirect',
331 400: '400 Bad Request',
332 401: '401 Unauthorized',
333 402: '402 Payment Required',
334 403: '403 Forbidden',
335 404: '404 Not Found',
336 405: '405 Method Not Allowed',
337 406: '406 Not Acceptable',
340 411: '411 Length Required',
341 412: '412 Precondition Failed',
342 413: '413 Payload Too Large',
343 414: '414 URI Too Long',
344 415: '415 Unsupported Media Type',
345 416: '416 Range Not Satisfiable',
346 417: '417 Expectation Failed',
347 418: "418 I'm a teapot",
348 429: '429 Too Many Requests',
349 431: '431 Request Header Fields Too Large',
350 451: '451 Unavailable For Legal Reasons',
351 500: '500 Internal Server Error',
352 501: '501 Not Implemented',
353 503: '503 Service Unavailable',
356 def _get_headers(response):
357 return list(response.headers)
359 def _get_content(response):
360 content = response.content
362 if isinstance(content, bytes):
365 if isinstance(content, str):
366 return (content.encode('utf-8'),)
371 def app(env, start_fn):
372 response = handler(_get_request_from_env(env))
374 start_fn(_get_status(response), _get_headers(response))
375 return _get_content(response)