/
cache_control.py
269 lines (215 loc) · 9.47 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
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
"""Internal utilities for determining cache expiration and other cache actions.
.. automodsumm:: requests_cache.cache_control
:classes-only:
:nosignatures:
.. automodsumm:: requests_cache.cache_control
:functions-only:
:nosignatures:
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from fnmatch import fnmatch
from logging import getLogger
from math import ceil
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Tuple, Union
from attr import define, field
from requests import PreparedRequest, Response
from ._utils import coalesce
if TYPE_CHECKING:
from .models import CachedResponse
__all__ = ['DO_NOT_CACHE', 'CacheActions']
# May be set by either headers or expire_after param to disable caching or disable expiration
DO_NOT_CACHE = 0
NEVER_EXPIRE = -1
# Supported Cache-Control directives
CACHE_DIRECTIVES = ['immutable', 'max-age', 'no-cache', 'no-store']
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 class that translates cache settings and headers into specific actions to take for a
given cache item. Actions include:
* Read from the cache
* Write to the cache
* Set cache expiration
* Add headers for conditional requests
If multiple sources provide an expiration time, they will be used in the following order of
precedence:
1. Cache-Control request headers
2. Cache-Control response headers (if enabled)
3. Per-request expiration
4. Per-URL expiration
5. Per-session expiration
See :ref:`headers` for more details about behavior.
"""
cache_control: bool = field(default=False)
cache_key: str = field(default=None)
expire_after: ExpirationTime = field(default=None)
request_directives: Dict[str, str] = field(factory=dict)
skip_read: bool = field(default=False)
skip_write: bool = field(default=False)
validation_headers: Dict[str, str] = field(factory=dict)
@classmethod
def from_request(
cls,
cache_key: str,
request: PreparedRequest,
cache_control: bool = False,
session_expire_after: ExpirationTime = None,
urls_expire_after: ExpirationPatterns = None,
request_expire_after: ExpirationTime = None,
**kwargs,
):
"""Initialize from request info and cache settings"""
directives = get_cache_directives(request.headers)
logger.debug(f'Cache directives from request headers: {directives}')
# Check expiration values in order of precedence
expire_after = coalesce(
directives.get('max-age'),
request_expire_after,
get_url_expiration(request.url, urls_expire_after),
session_expire_after,
)
# Check conditions for caching based on request headers. Also check expire_after options
# unless cache_control=True, in which case these may be overridden by response headers.
check_expiration = directives.get('max-age') if cache_control else expire_after
skip_write = check_expiration == DO_NOT_CACHE or 'no-store' in directives
return cls(
cache_control=cache_control,
cache_key=cache_key,
expire_after=expire_after,
request_directives=directives,
skip_read=skip_write or 'no-cache' in directives,
skip_write=skip_write,
)
@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_cached_response(self, response: CachedResponse):
"""Check for relevant cache headers from a cached response, and set headers for a
conditional request, if possible.
Used after fetching a cached response, but before potentially sending a new request
(if expired).
"""
if not response or not response.is_expired:
return
if response.headers.get('ETag'):
self.validation_headers['If-None-Match'] = response.headers['ETag']
if response.headers.get('Last-Modified'):
self.validation_headers['If-Modified-Since'] = response.headers['Last-Modified']
def update_from_response(self, response: Response):
"""Update expiration + actions based on headers from a new response.
Used after receiving a new response but before saving it to the cache.
"""
if not response or not self.cache_control:
return
directives = get_cache_directives(response.headers)
logger.debug(f'Cache directives from response headers: {directives}')
# Check headers for expiration, validators, and other cache directives
if directives.get('immutable'):
self.expire_after = NEVER_EXPIRE
else:
self.expire_after = coalesce(
directives.get('max-age'), directives.get('expires'), self.expire_after
)
has_validator = response.headers.get('ETag') or response.headers.get('Last-Modified')
no_store = 'no-store' in directives or 'no-store' in self.request_directives
# If expiration is 0 and there's a validator, save it to the cache and revalidate on use
# Otherwise, skip writing to the cache if specified by expiration or other headers
expire_immediately = try_int(self.expire_after) == DO_NOT_CACHE
self.skip_write = (expire_immediately or no_store) and not has_validator
def get_expiration_datetime(expire_after: ExpirationTime) -> Optional[datetime]:
"""Convert an expiration value in any supported format to an absolute datetime"""
# Never expire
if expire_after is None or expire_after == NEVER_EXPIRE:
return None
# Expire immediately
elif try_int(expire_after) == DO_NOT_CACHE:
return datetime.utcnow()
# Already a datetime or datetime str
if isinstance(expire_after, str):
return parse_http_date(expire_after)
elif isinstance(expire_after, datetime):
return to_utc(expire_after)
# Otherwise, it must be a timedelta or time in seconds
if not isinstance(expire_after, timedelta):
expire_after = timedelta(seconds=expire_after)
return datetime.utcnow() + expire_after
def get_expiration_seconds(expire_after: ExpirationTime) -> int:
"""Convert an expiration value in any supported format to an expiration time in seconds"""
expires = get_expiration_datetime(expire_after)
return ceil((expires - datetime.utcnow()).total_seconds()) if expires else NEVER_EXPIRE
def get_cache_directives(headers: Mapping) -> Dict:
"""Get all Cache-Control directives, and handle multiple headers and comma-separated lists"""
if not headers:
return {}
kv_directives = {}
if headers.get('Cache-Control'):
cache_directives = headers['Cache-Control'].split(',')
kv_directives = dict([split_kv_directive(value) for value in cache_directives])
if 'Expires' in headers:
kv_directives['expires'] = headers['Expires']
return kv_directives
def get_url_expiration(
url: Optional[str], urls_expire_after: ExpirationPatterns = None
) -> ExpirationTime:
"""Check for a matching per-URL expiration, if any"""
if not url:
return None
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 parse_http_date(value: str) -> Optional[datetime]:
"""Attempt to parse an HTTP (RFC 5322-compatible) timestamp"""
try:
expire_after = parsedate_to_datetime(value)
return to_utc(expire_after)
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 = dt.astimezone(timezone.utc)
dt = dt.replace(tzinfo=None)
return dt
def try_int(value: Any) -> Optional[int]:
"""Convert a value to an int, if possible, otherwise ``None``"""
try:
return int(value)
except (TypeError, ValueError):
return None
def url_match(url: str, 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
"""
url = url.split('://')[-1]
pattern = pattern.split('://')[-1].rstrip('*') + '**'
return fnmatch(url, pattern)