6 _Request = collections.namedtuple(
26 class Request(_Request):
27 def __new__(cls, env, **kwargs):
30 accept = env.get('HTTP_ACCEPT')
31 accept_encoding = env.get('HTTP_ACCEPT_ENCODING')
32 accept_language = env.get('HTTP_ACCEPT_LANGUAGE')
33 content = env.get('CONTENT', '')
34 content_type = env.get('CONTENT_TYPE')
35 method = env.get('REQUEST_METHOD')
36 path = env.get('PATH_INFO')
37 query = env.get('QUERY_STRING')
38 user_agent = env.get('HTTP_USER_AGENT')
40 content_length = env.get('CONTENT_LENGTH')
42 if content_length == '' or content_length is None:
46 content_length = int(content_length)
48 errors.append('Unable to parse Content-Length "{}"'.format(content_length))
52 cookie = http.cookies.SimpleCookie(env.get('HTTP_COOKIE'))
54 cookie = http.cookies.SimpleCookie()
58 GET = urllib.parse.parse_qs(query)
61 errors.append('Unable to parse GET parameters from query string "{}"'.format(query))
65 elif method == 'POST':
66 raise Exception('not yet implemented')
70 result = super().__new__(
75 accept_encoding=accept_encoding,
76 accept_language=accept_language,
78 content_length = content_length,
79 content_type = content_type,
82 parameters=parameters,
85 user_agent=user_agent,
88 if path.startswith('/'):
89 result.subpath = path[1:]
95 def _get_request_from_env(env):
98 _Response = collections.namedtuple(
108 class Response(_Response):
109 def __new__(cls, content, **kwargs):
110 status = kwargs.pop('status', 200)
111 assert isinstance(status, int)
113 content_type = kwargs.pop('content_type')
114 assert isinstance(content_type, str)
116 extra_headers = kwargs.pop('extra_headers', ())
117 assert isinstance(extra_headers, tuple)
119 assert len(kwargs) == 0
121 return super().__new__(
124 content_type=content_type,
125 extra_headers=extra_headers,
132 ('Content-Type', self.content_type),
135 class HTMLResponse(Response):
136 def __new__(cls, content, **kwargs):
137 assert 'content_type' not in kwargs
139 return super().__new__(
142 content_type='text/html',
146 class JSONResponse(Response):
147 def __new__(cls, content_json, **kwargs):
148 assert 'content_type' not in kwargs
149 assert 'content' not in kwargs
151 self = super().__new__(
153 content=json.dumps(content_json),
154 content_type='application/json',
157 self.content_json = content_json
160 class TextResponse(Response):
161 def __new__(cls, content, **kwargs):
162 assert 'content_type' not in kwargs
164 return super().__new__(
167 content_type='text/plain',
171 _RedirectResponse = collections.namedtuple(
179 class RedirectResponse(_RedirectResponse):
180 def __new__(cls, location, **kwargs):
181 assert isinstance(location, str)
183 permanent = kwargs.pop('permanent', True)
184 assert isinstance(permanent, bool)
185 assert len(kwargs) == 0
187 return super().__new__(
195 return 308 if self.permanent else 307
199 return (('Location', self.location),)
205 def default_file_not_found_handler(request):
206 return Response('', status=404)
208 def route_on_subpath(**kwargs):
209 routes = kwargs.pop('routes')
210 file_not_found_handler = kwargs.pop(
211 'file_not_found_hanlder',
212 default_file_not_found_handler,
216 raise Exception('Keyword argument "routes" is required')
219 raise Exception('Unexpected keyword argument')
221 def wrapped(request):
222 split_subpath = request.subpath.split('/', 1)
223 subpath = split_subpath[0]
225 if len(split_subpath) == 2:
226 request.subpath = split_subpath[1]
230 return routes.get(subpath, file_not_found_handler)(request)
246 def default_method_not_allowed_handler(request):
249 def default_options_handler(handlers):
250 def handler(request):
251 return Response(','.join(handlers.keys()))
254 def route_on_method(**kwargs):
256 for method in REQUEST_METHODS:
258 handlers[method] = kwargs.pop(method)
260 method_not_allowed_handler = kwargs.pop(
261 'method_not_allowed',
262 default_method_not_allowed_handler,
265 assert len(kwargs) == 0
267 if 'OPTIONS' not in handlers:
268 handlers['OPTIONS'] = default_options_handler(handlers)
270 def handler(request):
272 request.method.upper(),
273 method_not_allowed_handler,
278 def _get_status(response):
281 307: '307 Temporary Redirect',
282 308: '308 Permanent Redirect',
285 def _get_headers(response):
286 return list(response.headers)
288 def _get_content(response):
289 content = response.content
291 if isinstance(content, bytes):
294 if isinstance(content, str):
295 return (content.encode('utf-8'),)
300 def app(env, start_fn):
301 response = handler(_get_request_from_env(env))
303 start_fn(_get_status(response), _get_headers(response))
304 return _get_content(response)