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://wsgi.readthedocs.io/en/latest/proposals-2.0.html}. 
 6  """ 
 7   
 8  __all__ = [] 
 9   
10  import warnings 
11   
12  from wsgitools.filters import CloseableIterator, CloseableList 
13   
14  warnings.warn("wsgitools.adapters will be removed", DeprecationWarning) 
15   
16  __all__.append("WSGI2to1Adapter") 
17 -class WSGI2to1Adapter(object): 
18      """Adapts an application with an interface that might somewhen be known as 
19      WSGI 2.0 to the WSGI 1.0 interface.""" 
20 -    def __init__(self, app): 
21          """app is an application with an interface that might somewhen be known 
22          as WSGI 2.0.""" 
23          self.app = app 
24 -    def __call__(self, environ, start_response): 
25          """WSGI 1.0 interface""" 
26          assert isinstance(environ, dict) 
27          status, headers, iterable = self.app(environ) 
28          assert isinstance(status, str) 
29          assert isinstance(headers, list) 
30          assert hasattr(iterable, "__iter__") 
31          start_response(status, headers) 
32          return iterable 
33   
34  __all__.append("WSGI1to2Adapter") 
35 -class WSGI1to2Adapter(object): 
36      """Adapts a WSGI 1.0 application to something that might somewhen be the 
37      WSGI 2.0 interface.""" 
38 -    def __init__(self, app): 
39          """@param app: is a WSGI 1.0 application""" 
40          self.app = app 
41 -    def __call__(self, environ): 
42          """some interface that might somewhen be known as WSGI 2.0""" 
43          assert isinstance(environ, dict) 
44          results = [None, None, []] 
45          def start_response(status, headers, exc_info=None): 
46              assert isinstance(status, str) 
47              assert isinstance(headers, list) 
48              results[0] = status 
49              results[1] = headers 
50              def write_callable(data): 
51                  results[2].append(data) 
52              return write_callable 
53          iterable = self.app(environ, start_response) 
54          assert hasattr(iterable, "__iter__") 
55          if not results[2]: 
56              return results[0], results[1], iterable 
57          if isinstance(iterable, list): 
58              # retaining .close attribute this way 
59              iterable[:0] = results[2] 
60              return results[0], results[1], iterable 
61          close_function = getattr(iterable, "close", None) 
62          iterable = iter(iterable) 
63          try: 
64              first = next(iterable) 
65          except StopIteration: 
66              return (results[0], results[1], 
67                      CloseableList(close_function, results[2])) 
68          results[2].append(first) 
69          return (results[0], results[1], 
70                  CloseableIterator(close_function, results[2], iterable)) 
71