summaryrefslogtreecommitdiff
path: root/wsgitools/adapters.py
blob: 86a1412c692d92ddccc3669df115b6f3ed59d15a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
"""
There are attempts to create a new version of the WSGI standard. These
classes try to adapt the current standard to something that eventually
works out to be the next version of WSGI. For more information see
U{http://www.wsgi.org/wsgi/WSGI_2.0}.
"""

__all__ = []

from wsgitools.filters import CloseableIterator, CloseableList

__all__.append("WSGI2to1Adapter")
class WSGI2to1Adapter:
    """Adapts an application with an interface that might somewhen be known as
    WSGI 2.0 to the WSGI 1.0 interface."""
    def __init__(self, app):
        """app is an application with an interface that might somewhen be known
        as WSGI 2.0."""
        self.app = app
    def __call__(self, environ, start_response):
        """WSGI 1.0 interface"""
        assert isinstance(environ, dict)
        status, headers, iterable = self.app(environ)
        assert isinstance(status, str)
        assert isinstance(headers, list)
        assert hasattr(iterable, "__iter__")
        start_response(status, headers)
        return iterable

__all__.append("WSGI1to2Adapter")
class WSGI1to2Adapter:
    """Adapts a WSGI 1.0 application to something that might somewhen be the
    WSGI 2.0 interface."""
    def __init__(self, app):
        """@param app: is a WSGI 1.0 application"""
        self.app = app
    def __call__(self, environ):
        """some interface that might somewhen be known as WSGI 2.0"""
        assert isinstance(environ, dict)
        results = [None, None, []]
        def start_response(status, headers, exc_info=None):
            assert isinstance(status, str)
            assert isinstance(headers, list)
            results[0] = status
            results[1] = headers
            def write_callable(data):
                results[2].append(data)
            return write_callable
        iterable = self.app(environ, start_response)
        assert hasattr(iterable, "__iter__")
        if isinstance(iterable, list):
            # retaining .close attribute this way
            iterable[:0] = results[2]
            return results[0], results[1], results[2]
        close_function = getattr(iterable, "close", None)
        iterable = iter(iterable)
        try:
            first = iterable.next()
        except StopIteration:
            return (results[0], results[1],
                    CloseableList(close_function, results[2]))
        results[2].append(first)
        return (results[0], results[1],
                CloseableIterator(close_function, results[2], iterable))