diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/celpy/adapter.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/celpy/adapter.py | 137 |
1 files changed, 137 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/celpy/adapter.py b/.venv/lib/python3.12/site-packages/celpy/adapter.py new file mode 100644 index 00000000..572b65ce --- /dev/null +++ b/.venv/lib/python3.12/site-packages/celpy/adapter.py @@ -0,0 +1,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}") |