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

   1 import collections
   2 import json
   3
   4 Request = collections.namedtuple(
   5     'Request',
   6     (
   7         'environ',
   8     )
   9 )
  10
  11 _Response = collections.namedtuple(
  12     'Response',
  13     (
  14         'status',
  15         'content_type',
  16         'extra_headers',
  17         'content',
  18     ),
  19 )
  20
  21 class Response(_Response):
  22     def __new__(cls, content, **kwargs):
  23         status = kwargs.pop('status', 200)
  24         assert isinstance(status, int)
  25
  26         content_type = kwargs.pop('content_type')
  27         assert isinstance(content_type, str)
  28
  29         extra_headers = kwargs.pop('extra_headers', ())
  30         assert isinstance(extra_headers, tuple)
  31
  32         assert len(kwargs) == 0
  33
  34         return super().__new__(
  35             cls,
  36             status=status,
  37             content_type=content_type,
  38             extra_headers=extra_headers,
  39             content=content,
  40         )
  41
  42     @property
  43     def headers(self):
  44         return (
  45             ('Content-Type', self.content_type),
  46         )
  47
  48 class HTMLResponse(Response):
  49     def __new__(cls, content, **kwargs):
  50         assert 'content_type' not in kwargs
  51
  52         return super().__new__(
  53             cls,
  54             content,
  55             content_type='text/html',
  56             **kwargs,
  57         )
  58
  59 class JSONResponse(Response):
  60     def __new__(cls, content_json, **kwargs):
  61         assert 'content_type' not in kwargs
  62         assert 'content' not in kwargs
  63
  64         self = super().__new__(
  65             cls,
  66             content=json.dumps(content_json),
  67             content_type='application/json',
  68             **kwargs,
  69         )
  70         self.content_json = content_json
  71         return self
  72
  73 class TextResponse(Response):
  74     def __new__(cls, content, **kwargs):
  75         assert 'content_type' not in kwargs
  76
  77         return super().__new__(
  78             cls,
  79             content,
  80             content_type='text/plain',
  81             **kwargs,
  82         )
  83
  84 _RedirectResponse = collections.namedtuple(
  85     'RedirectResponse',
  86     (
  87         'location',
  88         'permanent',
  89     ),
  90 )
  91
  92 class RedirectResponse(_RedirectResponse):
  93     def __new__(cls, location, **kwargs):
  94         assert isinstance(location, str)
  95
  96         permanent = kwargs.pop('permanent', True)
  97         assert isinstance(permanent, bool)
  98         assert len(kwargs) == 0
  99
 100         return super().__new__(
 101             cls,
 102             location=location,
 103             permanent=permanent,
 104         )
 105
 106     @property
 107     def status(self):
 108         return 308 if self.permanent else 307
 109
 110     @property
 111     def headers(self):
 112         return (('Location', self.location),)
 113
 114     @property
 115     def content(self):
 116         return (b'',)
 117
 118 def _get_status(response):
 119     return {
 120         200: '200 OK',
 121         307: '307 Temporary Redirect',
 122         308: '308 Permanent Redirect',
 123     }[response.status]
 124
 125 def _get_headers(response):
 126     return list(response.headers)
 127
 128 def _get_content(response):
 129     content = response.content
 130
 131     if isinstance(content, bytes):
 132         return (content,)
 133
 134     if isinstance(content, str):
 135         return (content.encode('utf-8'),)
 136
 137     return content
 138
 139 def App(handler):
 140     def app(environ, start_fn):
 141         response = handler(Request(environ))
 142
 143         start_fn(_get_status(response), _get_headers(response))
 144         return _get_content(response)
 145     return app