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
|
"""
Utilities to break down :py:class:`.CachedResponse` objects into a dict of python builtin types
using `cattrs <https://cattrs.readthedocs.io>`_. This does the majority of the work needed for any
serialization format.
.. automodsumm:: requests_cache.serializers.cattrs
:classes-only:
:nosignatures:
.. automodsumm:: requests_cache.serializers.cattrs
:functions-only:
:nosignatures:
"""
from datetime import datetime, timedelta
from typing import Callable, Dict, ForwardRef, MutableMapping
from cattr import GenConverter
from requests.cookies import RequestsCookieJar, cookiejar_from_dict
from requests.structures import CaseInsensitiveDict
from urllib3.response import HTTPHeaderDict
from ..models import CachedResponse
from .pipeline import Stage
class CattrStage(Stage):
"""Base serializer class that does pre/post-processing with ``cattrs``. This can be used either
on its own, or as a stage within a :py:class:`.SerializerPipeline`.
"""
def __init__(self, factory: Callable[..., GenConverter] = None):
self.converter = init_converter(factory)
def dumps(self, value: CachedResponse) -> Dict:
if not isinstance(value, CachedResponse):
return value
return self.converter.unstructure(value)
def loads(self, value: Dict) -> CachedResponse:
if not isinstance(value, MutableMapping):
return value
return self.converter.structure(value, cl=CachedResponse)
def init_converter(factory: Callable[..., GenConverter] = None):
"""Make a converter to structure and unstructure nested objects within a :py:class:`.CachedResponse`"""
factory = factory or GenConverter
converter = factory(omit_if_default=True)
# Convert datetimes to and from iso-formatted strings
converter.register_unstructure_hook(datetime, lambda obj: obj.isoformat() if obj else None) # type: ignore
converter.register_structure_hook(datetime, _to_datetime)
# Convert timedeltas to and from float values in seconds
converter.register_unstructure_hook(timedelta, lambda obj: obj.total_seconds() if obj else None) # type: ignore
converter.register_structure_hook(timedelta, _to_timedelta)
# Convert dict-like objects to and from plain dicts
converter.register_unstructure_hook(RequestsCookieJar, lambda obj: dict(obj.items())) # type: ignore
converter.register_structure_hook(RequestsCookieJar, lambda obj, cls: cookiejar_from_dict(obj))
converter.register_unstructure_hook(CaseInsensitiveDict, dict)
converter.register_structure_hook(CaseInsensitiveDict, lambda obj, cls: CaseInsensitiveDict(obj))
converter.register_unstructure_hook(HTTPHeaderDict, dict)
converter.register_structure_hook(HTTPHeaderDict, lambda obj, cls: HTTPHeaderDict(obj))
# Tell cattrs that a 'CachedResponse' forward ref is equivalent to the CachedResponse class
converter.register_structure_hook(
ForwardRef('CachedResponse'),
lambda obj, cls: converter.structure(obj, CachedResponse),
)
return converter
def _to_datetime(obj, cls) -> datetime:
if isinstance(obj, str):
obj = datetime.fromisoformat(obj)
return obj
def _to_timedelta(obj, cls) -> timedelta:
if isinstance(obj, (int, float)):
obj = timedelta(seconds=obj)
return obj
|
