Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# The MIT License (MIT)
2#
3# Copyright (c) 2019 Looker Data Sciences, Inc.
4#
5# Permission is hereby granted, free of charge, to any person obtaining a copy
6# of this software and associated documentation files (the "Software"), to deal
7# in the Software without restriction, including without limitation the rights
8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9# copies of the Software, and to permit persons to whom the Software is
10# furnished to do so, subject to the following conditions:
11#
12# The above copyright notice and this permission notice shall be included in
13# all copies or substantial portions of the Software.
14#
15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21# THE SOFTWARE.
23"""Deserialize API response into models"""
25import datetime
26import enum
27import functools
28import json
29import keyword
30import sys
31from typing import (
32 Callable,
33 MutableMapping,
34 Sequence,
35 Type,
36 Union,
37)
39import cattr
41try:
42 from cattrs.cols import is_sequence, list_structure_factory
43except ImportError:
44 is_sequence = None # type: ignore
45 list_structure_factory = None # type: ignore
47from looker_sdk.rtl import model, hooks
50class DeserializeError(Exception):
51 """Improperly formatted data to deserialize."""
54TModelOrSequence = Union[
55 MutableMapping[str, str],
56 Sequence[int],
57 Sequence[str],
58 model.Model,
59 Sequence[model.Model],
60]
61TDeserializeReturn = TModelOrSequence
62TStructure = Union[Type[Sequence[int]], Type[Sequence[str]], Type[TDeserializeReturn]]
63TDeserialize = Callable[[str, TStructure], TDeserializeReturn]
64TSerialize = Callable[[TModelOrSequence], bytes]
67def deserialize(
68 *, data: str, structure: TStructure, converter: cattr.Converter
69) -> TDeserializeReturn:
70 """Translate API data into models."""
71 try:
72 data = json.loads(data)
73 except json.JSONDecodeError as ex:
74 raise DeserializeError(f"Bad json {ex}")
75 try:
76 if is_sequence is not None and list_structure_factory is not None:
77 converter.register_structure_hook_factory(
78 is_sequence, list_structure_factory
79 )
80 response: TDeserializeReturn = converter.structure( # type: ignore
81 data, structure
82 )
83 except (TypeError, AttributeError) as ex:
84 raise DeserializeError(f"Bad data {ex}")
85 return response
88def serialize(*, api_model: TModelOrSequence, converter: cattr.Converter) -> bytes:
89 """Translate api_model into formdata encoded json bytes"""
90 data = converter.unstructure(api_model) # type: ignore
91 return json.dumps(data, default=lambda o: o.__dict__).encode("utf-8") # type: ignore
94def forward_ref_structure_hook(context, converter, data, forward_ref):
95 """Applied to ForwardRef model and enum annotations
97 - Map reserved words in json keys to approriate (safe) names in model.
98 - handle ForwardRef types until github.com/Tinche/cattrs/pull/42/ is fixed
99 Note: this is the reason we need a "context" param and have to use a
100 partial func to register the hook. Once the issue is resolved we can
101 remove "context" and the partial.
102 """
103 data = hooks.tr_data_keys(data)
104 actual_type = eval(forward_ref.__forward_arg__, context, locals())
105 if issubclass(actual_type, enum.Enum):
106 instance = converter.structure(data, actual_type)
107 elif issubclass(actual_type, model.Model):
108 # cannot use converter.structure - recursion error
109 instance = converter.structure_attrs_fromdict(data, actual_type)
110 else:
111 raise DeserializeError(f"Unknown type to deserialize: {actual_type}")
112 return instance
115def translate_keys_structure_hook(converter, data, model_type):
116 """Applied only to models.Model"""
117 new_data = hooks.tr_data_keys(data)
118 ret = converter.structure_attrs_fromdict(new_data, model_type)
119 return ret
122converter40 = cattr.Converter()
123deserialize40 = functools.partial(deserialize, converter=converter40)
124serialize40 = functools.partial(serialize, converter=converter40)
126converter40.register_structure_hook(datetime.datetime, hooks.datetime_structure_hook)
127unstructure_hook40 = functools.partial(hooks.unstructure_hook, converter40) # type: ignore
128converter40.register_unstructure_hook(model.Model, unstructure_hook40) # type: ignore
129converter40.register_unstructure_hook(
130 datetime.datetime, hooks.datetime_unstructure_hook # type: ignore
131)