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

   1 import collections
   2 import json
   3 import urllib.parse
   4
   5 _Request = collections.namedtuple(
   6     'Request',
   7     (
   8         'env',
   9         'GET',
  10         'accept',
  11         'accept_encoding',
  12         'accept_language',
  13         'content',
  14         'content_length',
  15         'content_type',
  16         'cookies',
  17         'method',
  18         'path',
  19         'parameters',
  20         'query',
  21         'user_agent',
  22     )
  23 )
  24
  25 class Request(_Request):
  26     def __new__(cls, env):
  27         accept = env.get('HTTP_ACCEPT')
  28         accept_encoding = env.get('HTTP_ACCEPT_ENCODING')
  29         accept_language = env.get('HTTP_ACCEPT_LANGUAGE')
  30         content = env.get('CONTENT', '')
  31         content_length = env.get('CONTENT_LENGTH')
  32         content_type = env.get('CONTENT_TYPE')
  33         cookies = env.get('HTTP_COOKIE')
  34         method = env.get('REQUEST_METHOD')
  35         path = env.get('PATH_INFO')
  36         query = env.get('QUERY_STRING')
  37         user_agent = env.get('HTTP_USER_AGENT')
  38
  39         GET = urllib.parse.parse_qs(query)
  40
  41         if method == 'GET':
  42             parameters = GET
  43
  44         result = super().__new__(
  45             cls,
  46             env=env,
  47             GET=GET,
  48             accept=accept,
  49             accept_encoding=accept_encoding,
  50             accept_language=accept_language,
  51             content = content,
  52             content_length = content_length,
  53             content_type = content_type,
  54             cookies=cookies,
  55             method=method,
  56             parameters=parameters,
  57             path=path,
  58             query=query,
  59             user_agent=user_agent,
  60         )
  61
  62         result.subpath = path
  63         return result
  64
  65 _Response = collections.namedtuple(
  66     'Response',
  67     (
  68         'status',
  69         'content_type',
  70         'extra_headers',
  71         'content',
  72     ),
  73 )
  74
  75 class Response(_Response):
  76     def __new__(cls, content, **kwargs):
  77         status = kwargs.pop('status', 200)
  78         assert isinstance(status, int)
  79
  80         content_type = kwargs.pop('content_type')
  81         assert isinstance(content_type, str)
  82
  83         extra_headers = kwargs.pop('extra_headers', ())
  84         assert isinstance(extra_headers, tuple)
  85
  86         assert len(kwargs) == 0
  87
  88         return super().__new__(
  89             cls,
  90             status=status,
  91             content_type=content_type,
  92             extra_headers=extra_headers,
  93             content=content,
  94         )
  95
  96     @property
  97     def headers(self):
  98         return (
  99             ('Content-Type', self.content_type),
 100         )
 101
 102 class HTMLResponse(Response):
 103     def __new__(cls, content, **kwargs):
 104         assert 'content_type' not in kwargs
 105
 106         return super().__new__(
 107             cls,
 108             content,
 109             content_type='text/html',
 110             **kwargs,
 111         )
 112
 113 class JSONResponse(Response):
 114     def __new__(cls, content_json, **kwargs):
 115         assert 'content_type' not in kwargs
 116         assert 'content' not in kwargs
 117
 118         self = super().__new__(
 119             cls,
 120             content=json.dumps(content_json),
 121             content_type='application/json',
 122             **kwargs,
 123         )
 124         self.content_json = content_json
 125         return self
 126
 127 class TextResponse(Response):
 128     def __new__(cls, content, **kwargs):
 129         assert 'content_type' not in kwargs
 130
 131         return super().__new__(
 132             cls,
 133             content,
 134             content_type='text/plain',
 135             **kwargs,
 136         )
 137
 138 _RedirectResponse = collections.namedtuple(
 139     'RedirectResponse',
 140     (
 141         'location',
 142         'permanent',
 143     ),
 144 )
 145
 146 class RedirectResponse(_RedirectResponse):
 147     def __new__(cls, location, **kwargs):
 148         assert isinstance(location, str)
 149
 150         permanent = kwargs.pop('permanent', True)
 151         assert isinstance(permanent, bool)
 152         assert len(kwargs) == 0
 153
 154         return super().__new__(
 155             cls,
 156             location=location,
 157             permanent=permanent,
 158         )
 159
 160     @property
 161     def status(self):
 162         return 308 if self.permanent else 307
 163
 164     @property
 165     def headers(self):
 166         return (('Location', self.location),)
 167
 168     @property
 169     def content(self):
 170         return (b'',)
 171
 172 def _get_status(response):
 173     return {
 174         200: '200 OK',
 175         307: '307 Temporary Redirect',
 176         308: '308 Permanent Redirect',
 177     }[response.status]
 178
 179 def _get_headers(response):
 180     return list(response.headers)
 181
 182 def _get_content(response):
 183     content = response.content
 184
 185     if isinstance(content, bytes):
 186         return (content,)
 187
 188     if isinstance(content, str):
 189         return (content.encode('utf-8'),)
 190
 191     return content
 192
 193 def App(handler):
 194     def app(env, start_fn):
 195         response = handler(Request(env))
 196
 197         start_fn(_get_status(response), _get_headers(response))
 198         return _get_content(response)
 199     return app