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 def _get_status(response):
  85     return {
  86         200: '200 OK',
  87     }[response.status]
  88
  89 def _get_headers(response):
  90     return list(response.headers)
  91
  92 def _get_content(response):
  93     content = response.content
  94
  95     if isinstance(content, bytes):
  96         return (content,)
  97
  98     if isinstance(content, str):
  99         return (content.encode('utf-8'),)
 100
 101     return content
 102
 103 def App(handler):
 104     def app(environ, start_fn):
 105         response = handler(Request(environ))
 106
 107         start_fn(_get_status(response), _get_headers(response))
 108         return _get_content(response)
 109     return app