Diff for "WSGI"

   1 def simple_app(environ, start_response):
   2     """Simplest possible application object"""
   3     status = '200 OK'
   4     response_headers = [('Content-type','text/plain')]
   5     start_response(status, response_headers)
   6     return ['Hello world!\n']
   7 
   8 
   9 class AppClass:
  10     """Produce the same output, but using a class
  11 
  12     (Note: 'AppClass' is the "application" here, so calling it
  13     returns an instance of 'AppClass', which is then the iterable
  14     return value of the "application callable" as required by
  15     the spec.
  16 
  17     If we wanted to use *instances* of 'AppClass' as application
  18     objects instead, we would have to implement a '__call__'
  19     method, which would be invoked to execute the application,
  20     and we would need to create an instance for use by the
  21     server or gateway.
  22     """
  23 
  24     def __init__(self, environ, start_response):
  25         self.environ = environ
  26         self.start = start_response
  27 
  28     def __iter__(self):
  29         status = '200 OK'
  30         response_headers = [('Content-type','text/plain')]
  31         self.start(status, response_headers)
  32         yield "Hello world!\n"

   1 import os, sys
   2 
   3 def run_with_cgi(application):
   4 
   5     environ = dict(os.environ.items())
   6     environ['wsgi.input']        = sys.stdin
   7     environ['wsgi.errors']       = sys.stderr
   8     environ['wsgi.version']      = (1,0)
   9     environ['wsgi.multithread']  = False
  10     environ['wsgi.multiprocess'] = True
  11     environ['wsgi.run_once']    = True
  12 
  13     if environ.get('HTTPS','off') in ('on','1'):
  14         environ['wsgi.url_scheme'] = 'https'
  15     else:
  16         environ['wsgi.url_scheme'] = 'http'
  17 
  18     headers_set = []
  19     headers_sent = []
  20 
  21     def write(data):
  22         if not headers_set:
  23              raise AssertionError("write() before start_response()")
  24 
  25         elif not headers_sent:
  26              # Before the first output, send the stored headers
  27              status, response_headers = headers_sent[:] = headers_set
  28              sys.stdout.write('Status: %s\r\n' % status)
  29              for header in response_headers:
  30                  sys.stdout.write('%s: %s\r\n' % header)
  31              sys.stdout.write('\r\n')
  32 
  33         sys.stdout.write(data)
  34         sys.stdout.flush()
  35 
  36     def start_response(status,response_headers,exc_info=None):
  37         if exc_info:
  38             try:
  39                 if headers_sent:
  40                     # Re-raise original exception if headers sent
  41                     raise exc_info[0], exc_info[1], exc_info[2]
  42             finally:
  43                 exc_info = None     # avoid dangling circular ref
  44         elif headers_set:
  45             raise AssertionError("Headers already set!")
  46 
  47         headers_set[:] = [status,response_headers]
  48         return write
  49 
  50     result = application(environ, start_response)
  51     try:
  52         for data in result:
  53             if data:    # don't send headers until body appears
  54                 write(data)
  55         if not headers_sent:
  56             write('')   # send headers now if body was empty
  57     finally:
  58         if hasattr(result,'close'):
  59             result.close()

   1 from piglatin import piglatin
   2 
   3 class LatinIter:
   4 
   5     """Transform iterated output to piglatin, if it's okay to do so
   6 
   7     Note that the "okayness" can change until the application yields
   8     its first non-empty string, so 'transform_ok' has to be a mutable
   9     truth value."""
  10 
  11     def __init__(self,result,transform_ok):
  12         if hasattr(result,'close'):
  13             self.close = result.close
  14         self._next = iter(result).next
  15         self.transform_ok = transform_ok
  16 
  17     def __iter__(self):
  18         return self
  19 
  20     def next(self):
  21         if self.transform_ok:
  22             return piglatin(self._next())
  23         else:
  24             return self._next()
  25 
  26 class Latinator:
  27 
  28     # by default, don't transform output
  29     transform = False
  30 
  31     def __init__(self, application):
  32         self.application = application
  33 
  34     def __call__(self, environ, start_response):
  35 
  36         transform_ok = []
  37 
  38         def start_latin(status,response_headers,exc_info=None):
  39 
  40             # Reset ok flag, in case this is a repeat call
  41             transform_ok[:]=[]
  42 
  43             for name,value in response_headers:
  44                 if name.lower()=='content-type' and value=='text/plain':
  45                     transform_ok.append(True)
  46                     # Strip content-length if present, else it'll be wrong
  47                     response_headers = [(name,value)
  48                         for name,value in response_headers
  49                             if name.lower()<>'content-length'
  50                     ]
  51                     break
  52 
  53             write = start_response(status,response_headers,exc_info)
  54 
  55             if transform_ok:
  56                 def write_latin(data):
  57                     write(piglatin(data))
  58                 return write_latin
  59             else:
  60                 return write
  61 
  62         return LatinIter(self.application(environ,start_latin),transform_ok)
  63 
  64 
  65 # Run foo_app under a Latinator's control, using the example CGI gateway
  66 from foo_app import foo_app
  67 run_with_cgi(Latinator(foo_app))