/
abc.py
148 lines (103 loc) · 3.32 KB
/
abc.py
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
import asyncio
import sys
from abc import ABC, abstractmethod
from collections.abc import Iterable, Sized
PY_35 = sys.version_info >= (3, 5)
class AbstractRouter(ABC):
def __init__(self):
self._frozen = False
def post_init(self, app):
"""Post init stage.
Not an abstract method for sake of backward compatibility,
but if the router wants to be aware of the application
it can override this.
"""
@property
def frozen(self):
return self._frozen
def freeze(self):
"""Freeze router."""
self._frozen = True
@asyncio.coroutine # pragma: no branch
@abstractmethod
def resolve(self, request):
"""Return MATCH_INFO for given request"""
class AbstractMatchInfo(ABC):
@asyncio.coroutine # pragma: no branch
@abstractmethod
def handler(self, request):
"""Execute matched request handler"""
@asyncio.coroutine # pragma: no branch
@abstractmethod
def expect_handler(self, request):
"""Expect handler for 100-continue processing"""
@property # pragma: no branch
@abstractmethod
def http_exception(self):
"""HTTPException instance raised on router's resolving, or None"""
@abstractmethod # pragma: no branch
def get_info(self):
"""Return a dict with additional info useful for introspection"""
@property # pragma: no branch
@abstractmethod
def apps(self):
"""Stack of nested applications.
Top level application is left-most element.
"""
@abstractmethod
def add_app(self, app):
"""Add application to the nested apps stack."""
@abstractmethod
def freeze(self):
"""Freeze the match info.
The method is called after route resolution.
After the call .add_app() is forbidden.
"""
class AbstractView(ABC):
def __init__(self, request):
self._request = request
@property
def request(self):
return self._request
@asyncio.coroutine # pragma: no branch
@abstractmethod
def __iter__(self):
while False: # pragma: no cover
yield None
if PY_35: # pragma: no branch
@abstractmethod
def __await__(self):
return # pragma: no cover
class AbstractResolver(ABC):
@asyncio.coroutine # pragma: no branch
@abstractmethod
def resolve(self, hostname):
"""Return IP address for given hostname"""
@asyncio.coroutine # pragma: no branch
@abstractmethod
def close(self):
"""Release resolver"""
class AbstractCookieJar(Sized, Iterable):
def __init__(self, *, loop=None):
self._loop = loop or asyncio.get_event_loop()
@abstractmethod
def clear(self):
"""Clear all cookies."""
@abstractmethod
def update_cookies(self, cookies, response_url=None):
"""Update cookies."""
@abstractmethod
def filter_cookies(self, request_url):
"""Return the jar's cookies filtered by their attributes."""
class AbstractPayloadWriter(ABC):
@abstractmethod
def write(self, chunk):
"""Write chunk into stream"""
@asyncio.coroutine
@abstractmethod
def write_eof(self, chunk=b''):
"""Write last chunk"""
@asyncio.coroutine
@abstractmethod
def drain(self):
"""Flush the write buffer."""