/
cache_control.py
244 lines (203 loc) · 8.67 KB
/
cache_control.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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
"""Utilities for determining cache expiration and other cache actions"""
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from fnmatch import fnmatch
from itertools import chain
from logging import getLogger
from typing import Any, Dict, Mapping, Optional, Tuple, Union
from aiohttp import ClientResponse
from aiohttp.typedefs import StrOrURL
from attr import define, field
from multidict import CIMultiDict
# Value that may be set by either Cache-Control headers or CacheBackend params to disable caching
DO_NOT_CACHE = 0
# Currently supported Cache-Control directives
CACHE_DIRECTIVES = ['max-age', 'no-cache', 'no-store']
# All cache-related headers, for logging/reference; not all are supported
REQUEST_CACHE_HEADERS = [
'Cache-Control',
'If-Unmodified-Since',
'If-Modified-Since',
'If-Match',
'If-None-Match',
]
RESPONSE_CACHE_HEADERS = ['Cache-Control', 'ETag', 'Expires', 'Age']
CacheDirective = Tuple[str, Union[None, int, bool]]
ExpirationTime = Union[None, int, float, str, datetime, timedelta]
ExpirationPatterns = Dict[str, ExpirationTime]
logger = getLogger(__name__)
@define()
class CacheActions:
"""A dataclass that contains info on specific actions to take for a given cache item.
This is determined by a combination of CacheBackend settings and request + response headers.
If multiple sources are provided, they will be used in the following order of precedence:
1. Cache-Control request headers (if enabled)
2. Cache-Control response headers (if enabled)
3. Per-request expiration
4. Per-URL expiration
5. Per-session expiration
"""
cache_control: bool = field(default=False)
expire_after: ExpirationTime = field(default=None)
key: str = field(default=None)
revalidate: bool = field(default=False) # Note: Revalidation is not currently implemented
skip_read: bool = field(default=False)
skip_write: bool = field(default=False)
@classmethod
def from_request(
cls,
key: str,
cache_control: bool = False,
headers: Mapping = None,
**kwargs,
):
"""Initialize from request info and CacheBackend settings"""
headers = headers or {}
if cache_control and has_cache_headers(headers):
return cls.from_headers(key, headers)
else:
return cls.from_settings(key, cache_control=cache_control, **kwargs)
@classmethod
def from_headers(cls, key: str, headers: Mapping):
"""Initialize from request headers"""
directives = get_cache_directives(headers)
do_not_cache = directives.get('max-age') == DO_NOT_CACHE
return cls(
cache_control=True,
key=key,
expire_after=directives.get('max-age'),
skip_read=do_not_cache or 'no-store' in directives or 'no-cache' in directives,
skip_write=do_not_cache or 'no-store' in directives,
revalidate=do_not_cache,
)
@classmethod
def from_settings(
cls,
key: str,
url: StrOrURL,
cache_control: bool = False,
request_expire_after: ExpirationTime = None,
session_expire_after: ExpirationTime = None,
urls_expire_after: ExpirationPatterns = None,
**kwargs,
):
"""Initialize from CacheBackend settings"""
# Check expire_after values in order of precedence
expire_after = coalesce(
request_expire_after,
get_url_expiration(url, urls_expire_after),
session_expire_after,
)
do_not_cache = expire_after == DO_NOT_CACHE
return cls(
cache_control=cache_control,
key=key,
expire_after=expire_after,
skip_read=do_not_cache,
skip_write=do_not_cache,
revalidate=do_not_cache,
)
@property
def expires(self) -> Optional[datetime]:
"""Convert the user/header-provided expiration value to a datetime"""
return get_expiration_datetime(self.expire_after)
def update_from_response(self, response: ClientResponse):
"""Update expiration + actions based on response headers, if not previously set by request"""
if not self.cache_control:
return
directives = get_cache_directives(response.headers)
do_not_cache = directives.get('max-age') == DO_NOT_CACHE
self.expire_after = coalesce(
self.expires, directives.get('max-age'), directives.get('expires')
)
self.skip_write = self.skip_write or do_not_cache or 'no-store' in directives
self.revalidate = self.revalidate or do_not_cache
def coalesce(*values: Any, default=None) -> Any:
"""Get the first non-``None`` value in a list of values"""
return next((v for v in values if v is not None), default)
def get_expiration_datetime(expire_after: ExpirationTime) -> Optional[datetime]:
"""Convert an expiration value in any supported format to an absolute datetime"""
logger.debug(f'Determining expiration time based on: {expire_after}')
if isinstance(expire_after, str):
expire_after = parse_http_date(expire_after)
if expire_after is None or expire_after == -1:
return None
if isinstance(expire_after, datetime):
return to_utc(expire_after)
if not isinstance(expire_after, timedelta):
assert isinstance(expire_after, (int, float))
expire_after = timedelta(seconds=expire_after)
return datetime.utcnow() + expire_after
def get_cache_directives(headers: Mapping) -> Dict:
"""Get all Cache-Control directives, and handle multiple headers and comma-separated lists"""
if not headers:
return {}
if not hasattr(headers, 'getall'):
headers = CIMultiDict(headers)
header_values = headers.getall('Cache-Control', []) # type: ignore
cache_directives = [v.split(',') for v in header_values if v]
cache_directives = list(chain.from_iterable(cache_directives))
kv_directives = dict([split_kv_directive(value) for value in cache_directives])
if 'Expires' in headers:
kv_directives['expires'] = headers.getone('Expires') # type: ignore
return kv_directives
def get_url_expiration(
url: StrOrURL, urls_expire_after: ExpirationPatterns = None
) -> ExpirationTime:
"""Check for a matching per-URL expiration, if any"""
for pattern, expire_after in (urls_expire_after or {}).items():
if url_match(url, pattern):
logger.debug(f'URL {url} matched pattern "{pattern}": {expire_after}')
return expire_after
return None
def has_cache_headers(headers: Mapping) -> bool:
"""Determine if headers contain cache directives **that we currently support**"""
headers = CIMultiDict(headers)
cache_control = ','.join(headers.getall('Cache-Control', []))
return any([d in cache_control for d in CACHE_DIRECTIVES] + [bool(headers.get('Expires'))])
def parse_http_date(value: str) -> Optional[datetime]:
"""Attempt to parse an HTTP (RFC 5322-compatible) timestamp"""
try:
return parsedate_to_datetime(value)
except (TypeError, ValueError):
logger.debug(f'Failed to parse timestamp: {value}')
return None
def split_kv_directive(header_value: str) -> CacheDirective:
"""Split a cache directive into a ``(header_value, int)`` key-value pair, if possible;
otherwise just ``(header_value, True)``.
"""
header_value = header_value.strip()
if '=' in header_value:
k, v = header_value.split('=', 1)
return k, try_int(v)
else:
return header_value, True
def to_utc(dt: datetime):
""" "All internal datetimes are UTC and timezone-naive. Convert any user/header-provided
datetimes to the same format.
"""
if dt.tzinfo:
dt.astimezone(timezone.utc)
dt = dt.replace(tzinfo=None)
return dt
def try_int(value: Optional[str]) -> Optional[int]:
"""Convert a string value to an int, if possible, otherwise ``None``"""
return int(str(value)) if str(value).isnumeric() else None
def url_match(url: StrOrURL, pattern: str) -> bool:
"""Determine if a URL matches a pattern
Args:
url: URL to test. Its base URL (without protocol) will be used.
pattern: Glob pattern to match against. A recursive wildcard will be added if not present
Example:
>>> url_match('https://httpbin.org/delay/1', 'httpbin.org/delay')
True
>>> url_match('https://httpbin.org/stream/1', 'httpbin.org/*/1')
True
>>> url_match('https://httpbin.org/stream/2', 'httpbin.org/*/1')
False
"""
if not url:
return False
url = str(url).split('://')[-1]
pattern = pattern.split('://')[-1].rstrip('*') + '**'
return fnmatch(url, pattern)