6 _Request = collections.namedtuple(
26 class Request(_Request):
27 def __new__(cls, env):
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.cookie.SimpleCookie(env.get('HTTP_COOKIE'))
54 cookie = http.cookie.SimpleCookie()
58 GET = urllib.parse.parse_qs(query)
61 errors.append('Unable to parse GET parameters from query string "{}"'.format(query))
66 result = super().__new__(
71 accept_encoding=accept_encoding,
72 accept_language=accept_language,
74 content_length = content_length,
75 content_type = content_type,
78 parameters=parameters,
81 user_agent=user_agent,
87 _Response = collections.namedtuple(
97 class Response(_Response):
98 def __new__(cls, content, **kwargs):
99 status = kwargs.pop('status', 200)
100 assert isinstance(status, int)
102 content_type = kwargs.pop('content_type')
103 assert isinstance(content_type, str)
105 extra_headers = kwargs.pop('extra_headers', ())
106 assert isinstance(extra_headers, tuple)
108 assert len(kwargs) == 0
110 return super().__new__(
113 content_type=content_type,
114 extra_headers=extra_headers,
121 ('Content-Type', self.content_type),
124 class HTMLResponse(Response):
125 def __new__(cls, content, **kwargs):
126 assert 'content_type' not in kwargs
128 return super().__new__(
131 content_type='text/html',
135 class JSONResponse(Response):
136 def __new__(cls, content_json, **kwargs):
137 assert 'content_type' not in kwargs
138 assert 'content' not in kwargs
140 self = super().__new__(
142 content=json.dumps(content_json),
143 content_type='application/json',
146 self.content_json = content_json
149 class TextResponse(Response):
150 def __new__(cls, content, **kwargs):
151 assert 'content_type' not in kwargs
153 return super().__new__(
156 content_type='text/plain',
160 _RedirectResponse = collections.namedtuple(
168 class RedirectResponse(_RedirectResponse):
169 def __new__(cls, location, **kwargs):
170 assert isinstance(location, str)
172 permanent = kwargs.pop('permanent', True)
173 assert isinstance(permanent, bool)
174 assert len(kwargs) == 0
176 return super().__new__(
184 return 308 if self.permanent else 307
188 return (('Location', self.location),)
194 def _get_status(response):
197 307: '307 Temporary Redirect',
198 308: '308 Permanent Redirect',
201 def _get_headers(response):
202 return list(response.headers)
204 def _get_content(response):
205 content = response.content
207 if isinstance(content, bytes):
210 if isinstance(content, str):
211 return (content.encode('utf-8'),)
216 def app(env, start_fn):
217 response = handler(Request(env))
219 start_fn(_get_status(response), _get_headers(response))
220 return _get_content(response)