# 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}")