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
|
# SPDX-Copyright: Copyright (c) Capital One Services, LLC
# SPDX-License-Identifier: Apache-2.0
# Copyright 2020 Capital One Services, LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and limitations under the License.
"""
Type Adapter to convert Python-native types into CEL structures.
Currently, Atomic Python objects have direct use of types in :mod:`celpy.celtypes`.
Non-Atomic Python objects are characterized by JSON and Protobuf
objects. This module has functions to convert JSON objects to CEL.
The protobuf decoder is TBD.
A more sophisticated type injection capability may be needed to permit
additional types or extensions to :mod:`celpy.celtypes`.
"""
import base64
import datetime
import json
from typing import Any, Dict, List, Union, cast
from celpy import celtypes
JSON = Union[Dict[str, Any], List[Any], bool, float, int, str, None]
class CELJSONEncoder(json.JSONEncoder):
"""
An Encoder to export CEL objects as JSON text.
This is **not** a reversible transformation. Some things are coerced to strings
without any more detailed type marker.
Specifically timestamps, durations, and bytes.
"""
@staticmethod
def to_python(
cel_object: celtypes.Value) -> Union[celtypes.Value, List[Any], Dict[Any, Any], bool]:
"""Recursive walk through the CEL object, replacing BoolType with native bool instances.
This lets the :py:mod:`json` module correctly represent the obects
with JSON ``true`` and ``false``.
This will also replace ListType and MapType with native ``list`` and ``dict``.
All other CEL objects will be left intact. This creates an intermediate hybrid
beast that's not quite a :py:class:`celtypes.Value` because a few things have been replaced.
"""
if isinstance(cel_object, celtypes.BoolType):
return True if cel_object else False
elif isinstance(cel_object, celtypes.ListType):
return [CELJSONEncoder.to_python(item) for item in cel_object]
elif isinstance(cel_object, celtypes.MapType):
return {
CELJSONEncoder.to_python(key): CELJSONEncoder.to_python(value)
for key, value in cel_object.items()
}
else:
return cel_object
def encode(self, cel_object: celtypes.Value) -> str:
"""
Override built-in encode to create proper Python :py:class:`bool` objects.
"""
return super().encode(CELJSONEncoder.to_python(cel_object))
def default(self, cel_object: celtypes.Value) -> JSON:
if isinstance(cel_object, celtypes.TimestampType):
return str(cel_object)
elif isinstance(cel_object, celtypes.DurationType):
return str(cel_object)
elif isinstance(cel_object, celtypes.BytesType):
return base64.b64encode(cel_object).decode("ASCII")
else:
return cast(JSON, super().default(cel_object))
class CELJSONDecoder(json.JSONDecoder):
"""
An Encoder to import CEL objects from JSON to the extent possible.
This does not handle non-JSON types in any form. Coercion from string
to TimestampType or DurationType or BytesType is handled by celtype
constructors.
"""
def decode(self, source: str, _w: Any = None) -> Any:
raw_json = super().decode(source)
return json_to_cel(raw_json)
def json_to_cel(document: JSON) -> celtypes.Value:
"""Convert parsed JSON object from Python to CEL to the extent possible.
It's difficult to distinguish strings which should be timestamps or durations.
::
>>> from pprint import pprint
>>> from celpy.adapter import json_to_cel
>>> doc = json.loads('["str", 42, 3.14, null, true, {"hello": "world"}]')
>>> cel = json_to_cel(doc)
>>> pprint(cel)
ListType([StringType('str'), IntType(42), DoubleType(3.14), None, BoolType(True), \
MapType({StringType('hello'): StringType('world')})])
"""
if isinstance(document, bool):
return celtypes.BoolType(document)
elif isinstance(document, float):
return celtypes.DoubleType(document)
elif isinstance(document, int):
return celtypes.IntType(document)
elif isinstance(document, str):
return celtypes.StringType(document)
elif document is None:
return None
elif isinstance(document, (tuple, List)):
return celtypes.ListType(
[json_to_cel(item) for item in document]
)
elif isinstance(document, Dict):
return celtypes.MapType(
{json_to_cel(key): json_to_cel(value) for key, value in document.items()}
)
elif isinstance(document, datetime.datetime):
return celtypes.TimestampType(document)
elif isinstance(document, datetime.timedelta):
return celtypes.DurationType(document)
else:
raise ValueError(f"unexpected type {type(document)} in JSON structure {document!r}")
|
