two version of R2R are hereHEADmaster

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