code.kerkeslager.com Git - fwx/blob - fwx.py

   1 import collections
   2 import http.cookies
   3 import json
   4 import urllib.parse
   5
   6 _Request = collections.namedtuple(
   7     'Request',
   8     (
   9         'env',
  10         'GET',
  11         'accept',
  12         'accept_encoding',
  13         'accept_language',
  14         'content',
  15         'content_length',
  16         'content_type',
  17         'cookie',
  18         'method',
  19         'path',
  20         'parameters',
  21         'query',
  22         'user_agent',
  23     )
  24 )
  25
  26 class Request(_Request):
  27     def __new__(cls, env):
  28         errors = []
  29
  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')
  39
  40         content_length = env.get('CONTENT_LENGTH')
  41
  42         if content_length == '' or content_length is None:
  43             content_length = 0
  44         else:
  45             try:
  46                 content_length = int(content_length)
  47             except ValueError:
  48                 errors.append('Unable to parse Content-Length "{}"'.format(content_length))
  49                 content_length = 0
  50
  51         try:
  52             cookie = http.cookies.SimpleCookie(env.get('HTTP_COOKIE'))
  53         except:
  54             cookie = http.cookies.SimpleCookie()
  55
  56
  57         try:
  58             GET = urllib.parse.parse_qs(query)
  59         except:
  60             GET = {}
  61             errors.append('Unable to parse GET parameters from query string "{}"'.format(query))
  62
  63         if method == 'GET':
  64             parameters = GET
  65
  66         result = super().__new__(
  67             cls,
  68             env=env,
  69             GET=GET,
  70             accept=accept,
  71             accept_encoding=accept_encoding,
  72             accept_language=accept_language,
  73             content = content,
  74             content_length = content_length,
  75             content_type = content_type,
  76             cookie=cookie,
  77             method=method,
  78             parameters=parameters,
  79             path=path,
  80             query=query,
  81             user_agent=user_agent,
  82         )
  83
  84         result.subpath = path
  85         return result
  86
  87 _Response = collections.namedtuple(
  88     'Response',
  89     (
  90         'status',
  91         'content_type',
  92         'extra_headers',
  93         'content',
  94     ),
  95 )
  96
  97 class Response(_Response):
  98     def __new__(cls, content, **kwargs):
  99         status = kwargs.pop('status', 200)
 100         assert isinstance(status, int)
 101
 102         content_type = kwargs.pop('content_type')
 103         assert isinstance(content_type, str)
 104
 105         extra_headers = kwargs.pop('extra_headers', ())
 106         assert isinstance(extra_headers, tuple)
 107
 108         assert len(kwargs) == 0
 109
 110         return super().__new__(
 111             cls,
 112             status=status,
 113             content_type=content_type,
 114             extra_headers=extra_headers,
 115             content=content,
 116         )
 117
 118     @property
 119     def headers(self):
 120         return (
 121             ('Content-Type', self.content_type),
 122         )
 123
 124 class HTMLResponse(Response):
 125     def __new__(cls, content, **kwargs):
 126         assert 'content_type' not in kwargs
 127
 128         return super().__new__(
 129             cls,
 130             content,
 131             content_type='text/html',
 132             **kwargs,
 133         )
 134
 135 class JSONResponse(Response):
 136     def __new__(cls, content_json, **kwargs):
 137         assert 'content_type' not in kwargs
 138         assert 'content' not in kwargs
 139
 140         self = super().__new__(
 141             cls,
 142             content=json.dumps(content_json),
 143             content_type='application/json',
 144             **kwargs,
 145         )
 146         self.content_json = content_json
 147         return self
 148
 149 class TextResponse(Response):
 150     def __new__(cls, content, **kwargs):
 151         assert 'content_type' not in kwargs
 152
 153         return super().__new__(
 154             cls,
 155             content,
 156             content_type='text/plain',
 157             **kwargs,
 158         )
 159
 160 _RedirectResponse = collections.namedtuple(
 161     'RedirectResponse',
 162     (
 163         'location',
 164         'permanent',
 165     ),
 166 )
 167
 168 class RedirectResponse(_RedirectResponse):
 169     def __new__(cls, location, **kwargs):
 170         assert isinstance(location, str)
 171
 172         permanent = kwargs.pop('permanent', True)
 173         assert isinstance(permanent, bool)
 174         assert len(kwargs) == 0
 175
 176         return super().__new__(
 177             cls,
 178             location=location,
 179             permanent=permanent,
 180         )
 181
 182     @property
 183     def status(self):
 184         return 308 if self.permanent else 307
 185
 186     @property
 187     def headers(self):
 188         return (('Location', self.location),)
 189
 190     @property
 191     def content(self):
 192         return (b'',)
 193
 194 REQUEST_METHODS = (
 195     'GET',
 196     'HEAD',
 197     'POST',
 198     'PUT',
 199     'PATCH',
 200     'DELETE',
 201     'CONNECT',
 202     'OPTIONS',
 203     'TRACE',
 204 )
 205
 206 def default_method_not_allowed_handler(request):
 207     return Response('')
 208
 209 def default_options_handler(handlers):
 210     def handler(request):
 211         return Response(','.join(handlers.keys()))
 212     return handler
 213
 214 def route_on_method(**kwargs):
 215     handlers = {}
 216     for method in REQUEST_METHODS:
 217         if method in kwargs:
 218             handlers[method] = kwargs.pop(method)
 219
 220     method_not_allowed_handler = kwargs.pop(
 221         'method_not_allowed',
 222         default_method_not_allowed_handler,
 223     )
 224
 225     assert len(kwargs) == 0
 226
 227     if 'OPTIONS' not in handlers:
 228         handlers['OPTIONS'] = default_options_handler(handlers)
 229
 230     def handler(request):
 231         return handlers.get(
 232             request.method.upper(),
 233             method_not_allowed_handler,
 234         )(request)
 235
 236     return handler
 237
 238 def _get_status(response):
 239     return {
 240         200: '200 OK',
 241         307: '307 Temporary Redirect',
 242         308: '308 Permanent Redirect',
 243     }[response.status]
 244
 245 def _get_headers(response):
 246     return list(response.headers)
 247
 248 def _get_content(response):
 249     content = response.content
 250
 251     if isinstance(content, bytes):
 252         return (content,)
 253
 254     if isinstance(content, str):
 255         return (content.encode('utf-8'),)
 256
 257     return content
 258
 259 def App(handler):
 260     def app(env, start_fn):
 261         response = handler(Request(env))
 262
 263         start_fn(_get_status(response), _get_headers(response))
 264         return _get_content(response)
 265     return app