Source Code for Module wsgitools.adapters

 1  """ 
 2  There are attempts to create a new version of the WSGI standard. These 
 3  classes try to adapt the current standard to something that eventually 
 4  works out to be the next version of WSGI. For more information see 
 5  U{http://www.wsgi.org/wsgi/WSGI_2.0}. 
 6  """ 
 7   
 8  __all__ = [] 
 9   
10  from wsgitools.filters import CloseableIterator, CloseableList 
11   
12  try: 
13      next 
14  except NameError: 
15 -    def next(it): 
16          return it.next() 
17   
18  __all__.append("WSGI2to1Adapter") 
19 -class WSGI2to1Adapter: 
20      """Adapts an application with an interface that might somewhen be known as 
21      WSGI 2.0 to the WSGI 1.0 interface.""" 
22 -    def __init__(self, app): 
23          """app is an application with an interface that might somewhen be known 
24          as WSGI 2.0.""" 
25          self.app = app 
26 -    def __call__(self, environ, start_response): 
27          """WSGI 1.0 interface""" 
28          assert isinstance(environ, dict) 
29          status, headers, iterable = self.app(environ) 
30          assert isinstance(status, str) 
31          assert isinstance(headers, list) 
32          assert hasattr(iterable, "__iter__") 
33          start_response(status, headers) 
34          return iterable 
35   
36  __all__.append("WSGI1to2Adapter") 
37 -class WSGI1to2Adapter: 
38      """Adapts a WSGI 1.0 application to something that might somewhen be the 
39      WSGI 2.0 interface.""" 
40 -    def __init__(self, app): 
41          """@param app: is a WSGI 1.0 application""" 
42          self.app = app 
43 -    def __call__(self, environ): 
44          """some interface that might somewhen be known as WSGI 2.0""" 
45          assert isinstance(environ, dict) 
46          results = [None, None, []] 
47          def start_response(status, headers, exc_info=None): 
48              assert isinstance(status, str) 
49              assert isinstance(headers, list) 
50              results[0] = status 
51              results[1] = headers 
52              def write_callable(data): 
53                  results[2].append(data) 
54              return write_callable 
55          iterable = self.app(environ, start_response) 
56          assert hasattr(iterable, "__iter__") 
57          if not results[2]: 
58              return results[0], results[1], iterable 
59          if isinstance(iterable, list): 
60              # retaining .close attribute this way 
61              iterable[:0] = results[2] 
62              return results[0], results[1], iterable 
63          close_function = getattr(iterable, "close", None) 
64          iterable = iter(iterable) 
65          try: 
66              first = next(iterable) 
67          except StopIteration: 
68              return (results[0], results[1], 
69                      CloseableList(close_function, results[2])) 
70          results[2].append(first) 
71          return (results[0], results[1], 
72                  CloseableIterator(close_function, results[2], iterable)) 
73