# 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.
"""
CEL Types: wrappers on Python types to provide CEL semantics.
This can be used by a Python module to work with CEL-friendly values and CEL results.
Examples of distinctions between CEL and Python:
- Unlike Python ``bool``, CEL :py:class:`BoolType` won't do some math.
- CEL has ``int64`` and ``uint64`` subclasses of integer. These have specific ranges and
raise :exc:`ValueError` errors on overflow.
CEL types will raise :exc:`ValueError` for out-of-range values and :exc:`TypeError`
for operations they refuse.
The :py:mod:`evaluation` module can capture these exceptions and turn them into result values.
This can permit the logic operators to quietly silence them via "short-circuiting".
In the normal course of events, CEL's evaluator may attempt operations between a
CEL exception result and an instance of one of CEL types.
We rely on this leading to an ordinary Python :exc:`TypeError` to be raised to propogate
the error. Or. A logic operator may discard the error object.
The :py:mod:`evaluation` module extends these types with it's own :exc:`CELEvalError` exception.
We try to keep that as a separate concern from the core operator implementations here.
We leverage Python features, which means raising exceptions when there is a problem.
Types
=============
See https://github.com/google/cel-go/tree/master/common/types
These are the Go type definitions that are used by CEL:
- BoolType
- BytesType
- DoubleType
- DurationType
- IntType
- ListType
- MapType
- NullType
- StringType
- TimestampType
- TypeType
- UintType
The above types are handled directly byt CEL syntax.
e.g., ``42`` vs. ``42u`` vs. ``"42"`` vs. ``b"42"`` vs. ``42.``.
We provide matching Python class names for each of these types. The Python type names
are subclasses of Python native types, allowing a client to transparently work with
CEL results. A Python host should be able to provide values to CEL that will be tolerated.
A type hint of ``Value`` unifies these into a common hint.
The CEL Go implementation also supports protobuf types:
- dpb.Duration
- tpb.Timestamp
- structpb.ListValue
- structpb.NullValue
- structpb.Struct
- structpb.Value
- wrapperspb.BoolValue
- wrapperspb.BytesValue
- wrapperspb.DoubleValue
- wrapperspb.FloatValue
- wrapperspb.Int32Value
- wrapperspb.Int64Value
- wrapperspb.StringValue
- wrapperspb.UInt32Value
- wrapperspb.UInt64Value
These types involve expressions like the following::
google.protobuf.UInt32Value{value: 123u}
In this case, the well-known protobuf name is directly visible as CEL syntax.
There's a ``google`` package with the needed definitions.
Type Provider
==============================
A type provider can be bound to the environment, this will support additional types.
This appears to be a factory to map names of types to type classes.
Run-time type binding is shown by a CEL expression like the following::
TestAllTypes{single_uint32_wrapper: 432u}
The ``TestAllTypes`` is a protobuf type added to the CEL run-time. The syntax
is defined by this syntax rule::
member_object : member "{" [fieldinits] "}"
The ``member`` is part of a type provider library,
either a standard protobuf definition or an extension. The field inits build
values for the protobuf object.
See https://github.com/google/cel-go/blob/master/test/proto3pb/test_all_types.proto
for the ``TestAllTypes`` protobuf definition that is registered as a type provider.
This expression will describes a Protobuf ``uint32`` object.
Type Adapter
=============
So far, it appears that a type adapter wraps existing Go or C++ types
with CEL-required methods. This seems like it does not need to be implemented
in Python.
Numeric Details
===============
Integer division truncates toward zero.
The Go definition of modulus::
// Mod returns the floating-point remainder of x/y.
// The magnitude of the result is less than y and its
// sign agrees with that of x.
https://golang.org/ref/spec#Arithmetic_operators
"Go has the nice property that -a/b == -(a/b)."
::
x y x / y x % y
5 3 1 2
-5 3 -1 -2
5 -3 -1 2
-5 -3 1 -2
Python definition::
The modulo operator always yields a result
with the same sign as its second operand (or zero);
the absolute value of the result is strictly smaller than
the absolute value of the second operand.
Here's the essential rule::
x//y * y + x%y == x
However. Python ``//`` truncates toward negative infinity. Go ``/`` truncates toward zero.
To get Go-like behavior, we need to use absolute values and restore the signs later.
::
x_sign = -1 if x < 0 else +1
go_mod = x_sign * (abs(x) % abs(y))
return go_mod
Timzone Details
===============
An implementation may have additional timezone names that must be injected into
th dateutil.gettz() processing.
For example, there may be the following sequence:
1. A lowercase match for an alias or an existing dateutil timezone.
2. A titlecase match for an existing dateutil timezone.
3. The fallback, which is a +/-HH:MM string.
.. TODO: Permit an extension into the timezone lookup.
"""
import datetime
import logging
import re
from functools import reduce, wraps
from math import fsum
from typing import (Any, Callable, Dict, Iterable, List, Mapping, NoReturn,
Optional, Sequence, Tuple, Type, TypeVar, Union, cast,
overload)
import dateutil.parser
import dateutil.tz
logger = logging.getLogger("celtypes")
Value = Union[
'BoolType',
'BytesType',
'DoubleType',
'DurationType',
'IntType',
'ListType',
'MapType',
None, # Used instead of NullType
'StringType',
'TimestampType',
'UintType',
]
# The domain of types used to build Annotations.
CELType = Union[
Type['BoolType'],
Type['BytesType'],
Type['DoubleType'],
Type['DurationType'],
Type['IntType'],
Type['ListType'],
Type['MapType'],
Callable[..., None], # Used instead of NullType
Type['StringType'],
Type['TimestampType'],
Type['TypeType'], # Used to mark Protobuf Type values
Type['UintType'],
Type['PackageType'],
Type['MessageType'],
]
def type_matched(method: Callable[[Any, Any], Any]) -> Callable[[Any, Any], Any]:
"""Decorates a method to assure the "other" value has the same type."""
@wraps(method)
def type_matching_method(self: Any, other: Any) -> Any:
if not(issubclass(type(other), type(self)) or issubclass(type(self), type(other))):
raise TypeError(f"no such overload: {self!r} {type(self)} != {other!r} {type(other)}")
return method(self, other)
return type_matching_method
def logical_condition(e: Value, x: Value, y: Value) -> Value:
"""
CEL e ? x : y operator.
Choose one of x or y. Exceptions in the unchosen expression are ignored.
Example::
2 / 0 > 4 ? 'baz' : 'quux'
is a "division by zero" error.
::
>>> logical_condition(
... BoolType(True), StringType("this"), StringType("Not That"))
StringType('this')
>>> logical_condition(
... BoolType(False), StringType("Not This"), StringType("that"))
StringType('that')
"""
if not isinstance(e, BoolType):
raise TypeError(f"Unexpected {type(e)} ? {type(x)} : {type(y)}")
result = x if e else y
logger.debug("logical_condition(%r, %r, %r) = %r", e, x, y, result)
return result
def logical_and(x: Value, y: Value) -> Value:
"""
Native Python has a left-to-right rule.
CEL && is commutative with non-Boolean values, including error objects.
"""
if not isinstance(x, BoolType) and not isinstance(y, BoolType):
raise TypeError(f"{type(x)} {x!r} and {type(y)} {y!r}")
elif not isinstance(x, BoolType) and isinstance(y, BoolType):
if y:
return x # whatever && true == whatever
else:
return y # whatever && false == false
elif isinstance(x, BoolType) and not isinstance(y, BoolType):
if x:
return y # true && whatever == whatever
else:
return x # false && whatever == false
else:
return BoolType(cast(BoolType, x) and cast(BoolType, y))
def logical_not(x: Value) -> Value:
"""
Native python `not` isn't fully exposed for CEL types.
"""
if isinstance(x, BoolType):
result = BoolType(not x)
else:
raise TypeError(f"not {type(x)}")
logger.debug("logical_not(%r) = %r", x, result)
return result
def logical_or(x: Value, y: Value) -> Value:
"""
Native Python has a left-to-right rule: (True or y) is True, (False or y) is y.
CEL || is commutative with non-Boolean values, including errors.
``(x || false)`` is ``x``, and ``(false || y)`` is ``y``.
Example 1::
false || 1/0 != 0
is a "no matching overload" error.
Example 2::
(2 / 0 > 3 ? false : true) || true
is a "True"
If the operand(s) are not BoolType, we'll create an TypeError that will become a CELEvalError.
"""
if not isinstance(x, BoolType) and not isinstance(y, BoolType):
raise TypeError(f"{type(x)} {x!r} or {type(y)} {y!r}")
elif not isinstance(x, BoolType) and isinstance(y, BoolType):
if y:
return y # whatever || true == true
else:
return x # whatever || false == whatever
elif isinstance(x, BoolType) and not isinstance(y, BoolType):
if x:
return x # true || whatever == true
else:
return y # false || whatever == whatever
else:
return BoolType(cast(BoolType, x) or cast(BoolType, y))
class BoolType(int):
"""
Native Python permits unary operators on Booleans.
For CEL, We need to prevent -false from working.
"""
def __new__(cls: Type['BoolType'], source: Any) -> 'BoolType':
if source is None:
return super().__new__(cls, 0)
elif isinstance(source, BoolType):
return source
elif isinstance(source, MessageType):
return super().__new__(
cls,
cast(int, source.get(StringType("value")))
)
else:
return super().__new__(cls, source)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({bool(self)})"
def __str__(self) -> str:
return str(bool(self))
def __neg__(self) -> NoReturn:
raise TypeError("no such overload")
def __hash__(self) -> int:
return super().__hash__()
class BytesType(bytes):
"""Python's bytes semantics are close to CEL."""
def __new__(
cls: Type['BytesType'],
source: Union[str, bytes, Iterable[int], 'BytesType', 'StringType'],
*args: Any,
**kwargs: Any
) -> 'BytesType':
if source is None:
return super().__new__(cls, b'')
elif isinstance(source, (bytes, BytesType)):
return super().__new__(cls, source)
elif isinstance(source, (str, StringType)):
return super().__new__(cls, source.encode('utf-8'))
elif isinstance(source, MessageType):
return super().__new__(
cls,
cast(bytes, source.get(StringType("value"))) # type: ignore [attr-defined]
)
elif isinstance(source, Iterable):
return super().__new__(
cls,
cast(Iterable[int], source)
)
else:
raise TypeError(f"Invalid initial value type: {type(source)}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
class DoubleType(float):
"""
Native Python permits mixed type comparisons, doing conversions as needed.
For CEL, we need to prevent mixed-type comparisons from working.
TODO: Conversions from string? IntType? UintType? DoubleType?
"""
def __new__(cls: Type['DoubleType'], source: Any) -> 'DoubleType':
if source is None:
return super().__new__(cls, 0)
elif isinstance(source, MessageType):
return super().__new__(
cls,
cast(float, source.get(StringType("value")))
)
else:
return super().__new__(cls, source)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
def __str__(self) -> str:
text = str(float(self))
return text
def __neg__(self) -> 'DoubleType':
return DoubleType(super().__neg__())
def __mod__(self, other: Any) -> NoReturn:
raise TypeError(
f"found no matching overload for '_%_' applied to '(double, {type(other)})'")
def __truediv__(self, other: Any) -> 'DoubleType':
if cast(float, other) == 0.0:
return DoubleType("inf")
else:
return DoubleType(super().__truediv__(other))
def __rmod__(self, other: Any) -> NoReturn:
raise TypeError(
f"found no matching overload for '_%_' applied to '({type(other)}, double)'")
def __rtruediv__(self, other: Any) -> 'DoubleType':
if self == 0.0:
return DoubleType("inf")
else:
return DoubleType(super().__rtruediv__(other))
@type_matched
def __eq__(self, other: Any) -> bool:
return super().__eq__(other)
@type_matched
def __ne__(self, other: Any) -> bool:
return super().__ne__(other)
def __hash__(self) -> int:
return super().__hash__()
IntOperator = TypeVar('IntOperator', bound=Callable[..., int])
def int64(operator: IntOperator) -> IntOperator:
"""Apply an operation, but assure the value is within the int64 range."""
@wraps(operator)
def clamped_operator(*args: Any, **kwargs: Any) -> int:
result: int = operator(*args, **kwargs)
if -(2**63) <= result < 2**63:
return result
raise ValueError("overflow")
return cast(IntOperator, clamped_operator)
class IntType(int):
"""
A version of int with overflow errors outside int64 range.
features/integer_math.feature:277 "int64_overflow_positive"
>>> IntType(9223372036854775807) + IntType(1)
Traceback (most recent call last):
...
ValueError: overflow
>>> 2**63
9223372036854775808
features/integer_math.feature:285 "int64_overflow_negative"
>>> -IntType(9223372036854775808) - IntType(1)
Traceback (most recent call last):
...
ValueError: overflow
>>> IntType(DoubleType(1.9))
IntType(2)
>>> IntType(DoubleType(-123.456))
IntType(-123)
"""
def __new__(
cls: Type['IntType'],
source: Any,
*args: Any,
**kwargs: Any
) -> 'IntType':
convert: Callable[..., int]
if source is None:
return super().__new__(cls, 0)
elif isinstance(source, IntType):
return source
elif isinstance(source, MessageType):
# Used by protobuf.
return super().__new__(
cls,
cast(int, source.get(StringType("value")))
)
elif isinstance(source, (float, DoubleType)):
convert = int64(round)
elif isinstance(source, TimestampType):
convert = int64(lambda src: src.timestamp())
elif isinstance(source, (str, StringType)) and source[:2] in {"0x", "0X"}:
convert = int64(lambda src: int(src[2:], 16))
elif isinstance(source, (str, StringType)) and source[:3] in {"-0x", "-0X"}:
convert = int64(lambda src: -int(src[3:], 16))
else:
# Must tolerate "-" as part of the literal.
# See https://github.com/google/cel-spec/issues/126
convert = int64(int)
return super().__new__(cls, convert(source))
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
def __str__(self) -> str:
text = str(int(self))
return text
@int64
def __neg__(self) -> 'IntType':
return IntType(super().__neg__())
@int64
def __add__(self, other: Any) -> 'IntType':
return IntType(super().__add__(cast(IntType, other)))
@int64
def __sub__(self, other: Any) -> 'IntType':
return IntType(super().__sub__(cast(IntType, other)))
@int64
def __mul__(self, other: Any) -> 'IntType':
return IntType(super().__mul__(cast(IntType, other)))
@int64
def __truediv__(self, other: Any) -> 'IntType':
other = cast(IntType, other)
self_sign = -1 if self < IntType(0) else +1
other_sign = -1 if other < IntType(0) else +1
go_div = self_sign * other_sign * (abs(self) // abs(other))
return IntType(go_div)
__floordiv__ = __truediv__
@int64
def __mod__(self, other: Any) -> 'IntType':
self_sign = -1 if self < IntType(0) else +1
go_mod = self_sign * (abs(self) % abs(cast(IntType, other)))
return IntType(go_mod)
@int64
def __radd__(self, other: Any) -> 'IntType':
return IntType(super().__radd__(cast(IntType, other)))
@int64
def __rsub__(self, other: Any) -> 'IntType':
return IntType(super().__rsub__(cast(IntType, other)))
@int64
def __rmul__(self, other: Any) -> 'IntType':
return IntType(super().__rmul__(cast(IntType, other)))
@int64
def __rtruediv__(self, other: Any) -> 'IntType':
other = cast(IntType, other)
self_sign = -1 if self < IntType(0) else +1
other_sign = -1 if other < IntType(0) else +1
go_div = self_sign * other_sign * (abs(other) // abs(self))
return IntType(go_div)
__rfloordiv__ = __rtruediv__
@int64
def __rmod__(self, other: Any) -> 'IntType':
left_sign = -1 if other < IntType(0) else +1
go_mod = left_sign * (abs(other) % abs(self))
return IntType(go_mod)
@type_matched
def __eq__(self, other: Any) -> bool:
return super().__eq__(other)
@type_matched
def __ne__(self, other: Any) -> bool:
return super().__ne__(other)
@type_matched
def __lt__(self, other: Any) -> bool:
return super().__lt__(other)
@type_matched
def __le__(self, other: Any) -> bool:
return super().__le__(other)
@type_matched
def __gt__(self, other: Any) -> bool:
return super().__gt__(other)
@type_matched
def __ge__(self, other: Any) -> bool:
return super().__ge__(other)
def __hash__(self) -> int:
return super().__hash__()
def uint64(operator: IntOperator) -> IntOperator:
"""Apply an operation, but assure the value is within the uint64 range."""
@wraps(operator)
def clamped_operator(*args: Any, **kwargs: Any) -> int:
result = operator(*args, **kwargs)
if 0 <= result < 2**64:
return result
raise ValueError("overflow")
return cast(IntOperator, clamped_operator)
class UintType(int):
"""
A version of int with overflow errors outside uint64 range.
Alternatives:
Option 1 - Use https://pypi.org/project/fixedint/
Option 2 - use array or struct modules to access an unsigned object.
Test Cases:
features/integer_math.feature:149 "unary_minus_no_overload"
>>> -UintType(42)
Traceback (most recent call last):
...
TypeError: no such overload
uint64_overflow_positive
>>> UintType(18446744073709551615) + UintType(1)
Traceback (most recent call last):
...
ValueError: overflow
uint64_overflow_negative
>>> UintType(0) - UintType(1)
Traceback (most recent call last):
...
ValueError: overflow
>>> - UintType(5)
Traceback (most recent call last):
...
TypeError: no such overload
"""
def __new__(
cls: Type['UintType'],
source: Any,
*args: Any,
**kwargs: Any
) -> 'UintType':
convert: Callable[..., int]
if isinstance(source, UintType):
return source
elif isinstance(source, (float, DoubleType)):
convert = uint64(round)
elif isinstance(source, TimestampType):
convert = uint64(lambda src: src.timestamp())
elif isinstance(source, (str, StringType)) and source[:2] in {"0x", "0X"}:
convert = uint64(lambda src: int(src[2:], 16))
elif isinstance(source, MessageType):
# Used by protobuf.
convert = uint64(lambda src: src['value'] if src['value'] is not None else 0)
elif source is None:
convert = uint64(lambda src: 0)
else:
convert = uint64(int)
return super().__new__(cls, convert(source))
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
def __str__(self) -> str:
text = str(int(self))
return text
def __neg__(self) -> NoReturn:
raise TypeError("no such overload")
@uint64
def __add__(self, other: Any) -> 'UintType':
return UintType(super().__add__(cast(IntType, other)))
@uint64
def __sub__(self, other: Any) -> 'UintType':
return UintType(super().__sub__(cast(IntType, other)))
@uint64
def __mul__(self, other: Any) -> 'UintType':
return UintType(super().__mul__(cast(IntType, other)))
@uint64
def __truediv__(self, other: Any) -> 'UintType':
return UintType(super().__floordiv__(cast(IntType, other)))
__floordiv__ = __truediv__
@uint64
def __mod__(self, other: Any) -> 'UintType':
return UintType(super().__mod__(cast(IntType, other)))
@uint64
def __radd__(self, other: Any) -> 'UintType':
return UintType(super().__radd__(cast(IntType, other)))
@uint64
def __rsub__(self, other: Any) -> 'UintType':
return UintType(super().__rsub__(cast(IntType, other)))
@uint64
def __rmul__(self, other: Any) -> 'UintType':
return UintType(super().__rmul__(cast(IntType, other)))
@uint64
def __rtruediv__(self, other: Any) -> 'UintType':
return UintType(super().__rfloordiv__(cast(IntType, other)))
__rfloordiv__ = __rtruediv__
@uint64
def __rmod__(self, other: Any) -> 'UintType':
return UintType(super().__rmod__(cast(IntType, other)))
@type_matched
def __eq__(self, other: Any) -> bool:
return super().__eq__(other)
@type_matched
def __ne__(self, other: Any) -> bool:
return super().__ne__(other)
def __hash__(self) -> int:
return super().__hash__()
class ListType(List[Value]):
"""
Native Python implements comparison operations between list objects.
For CEL, we prevent list comparison operators from working.
We provide an :py:meth:`__eq__` and :py:meth:`__ne__` that
gracefully ignore type mismatch problems, calling them not equal.
See https://github.com/google/cel-spec/issues/127
An implied logical And means a singleton behaves in a distinct way from a non-singleton list.
"""
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
def __lt__(self, other: Any) -> NoReturn:
raise TypeError("no such overload")
def __le__(self, other: Any) -> NoReturn:
raise TypeError("no such overload")
def __gt__(self, other: Any) -> NoReturn:
raise TypeError("no such overload")
def __ge__(self, other: Any) -> NoReturn:
raise TypeError("no such overload")
def __eq__(self, other: Any) -> bool:
if not isinstance(other, (list, ListType)):
raise TypeError(f"no such overload: ListType == {type(other)}")
def equal(s: Any, o: Any) -> Value:
try:
return BoolType(s == o)
except TypeError as ex:
return cast(BoolType, ex) # Instead of Union[BoolType, TypeError]
result = (
len(self) == len(other)
and reduce( # noqa: W503
logical_and, # type: ignore [arg-type]
(equal(item_s, item_o) for item_s, item_o in zip(self, other)),
BoolType(True) # type: ignore [arg-type]
)
)
if isinstance(result, TypeError):
raise result
return bool(result)
def __ne__(self, other: Any) -> bool:
if not isinstance(other, (list, ListType)):
raise TypeError(f"no such overload: ListType != {type(other)}")
def not_equal(s: Any, o: Any) -> Value:
try:
return BoolType(s != o)
except TypeError as ex:
return cast(BoolType, ex) # Instead of Union[BoolType, TypeError]
result = (
len(self) != len(other)
or reduce( # noqa: W503
logical_or, # type: ignore [arg-type]
(not_equal(item_s, item_o) for item_s, item_o in zip(self, other)),
BoolType(False) # type: ignore [arg-type]
)
)
if isinstance(result, TypeError):
raise result
return bool(result)
BaseMapTypes = Union[
Mapping[Any, Any],
Sequence[Tuple[Any, Any]],
None
]
MapKeyTypes = Union[
'IntType', 'UintType', 'BoolType', 'StringType', str
]
class MapType(Dict[Value, Value]):
"""
Native Python allows mapping updates and any hashable type as a kay.
CEL prevents mapping updates and has a limited domain of key types.
int, uint, bool, or string keys
We provide an :py:meth:`__eq__` and :py:meth:`__ne__` that
gracefully ignore type mismatch problems for the values, calling them not equal.
See https://github.com/google/cel-spec/issues/127
An implied logical And means a singleton behaves in a distinct way from a non-singleton mapping.
"""
def __init__(
self,
items: BaseMapTypes = None) -> None:
super().__init__()
if items is None:
pass
elif isinstance(items, Sequence):
for name, value in items:
self[name] = value
elif isinstance(items, Mapping):
for name, value in items.items():
self[name] = value
else:
raise TypeError(f"Invalid initial value type: {type(items)}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
def __getitem__(self, key: Any) -> Any:
if not MapType.valid_key_type(key):
raise TypeError(f"unsupported key type: {type(key)}")
return super().__getitem__(key)
def __eq__(self, other: Any) -> bool:
if not isinstance(other, (Mapping, MapType)):
raise TypeError(f"no such overload: MapType == {type(other)}")
def equal(s: Any, o: Any) -> BoolType:
try:
return BoolType(s == o)
except TypeError as ex:
return cast(BoolType, ex) # Instead of Union[BoolType, TypeError]
keys_s = self.keys()
keys_o = other.keys()
result = (
keys_s == keys_o
and reduce( # noqa: W503
logical_and, # type: ignore [arg-type]
(equal(self[k], other[k]) for k in keys_s),
BoolType(True) # type: ignore [arg-type]
)
)
if isinstance(result, TypeError):
raise result
return bool(result)
def __ne__(self, other: Any) -> bool:
if not isinstance(other, (Mapping, MapType)):
raise TypeError(f"no such overload: MapType != {type(other)}")
# Singleton special case, may return no-such overload.
if len(self) == 1 and len(other) == 1 and self.keys() == other.keys():
k = next(iter(self.keys()))
return cast(bool, self[k] != other[k]) # Instead of Union[BoolType, TypeError]
def not_equal(s: Any, o: Any) -> BoolType:
try:
return BoolType(s != o)
except TypeError as ex:
return cast(BoolType, ex) # Instead of Union[BoolType, TypeError]
keys_s = self.keys()
keys_o = other.keys()
result = (
keys_s != keys_o
or reduce( # noqa: W503
logical_or, # type: ignore [arg-type]
(not_equal(self[k], other[k]) for k in keys_s),
BoolType(False) # type: ignore [arg-type]
)
)
if isinstance(result, TypeError):
raise result
return bool(result)
@staticmethod
def valid_key_type(key: Any) -> bool:
"""Valid CEL key types. Plus native str for tokens in the source when evaluating ``e.f``"""
return isinstance(key, (IntType, UintType, BoolType, StringType, str))
class NullType:
"""TBD. May not be needed. Python's None semantics appear to match CEL perfectly."""
pass
class StringType(str):
"""Python's str semantics are very, very close to CEL.
We rely on the overlap between ``"/u270c"`` and ``"/U0001f431"`` in CEL and Python.
"""
def __new__(
cls: Type['StringType'],
source: Union[str, bytes, 'BytesType', 'StringType'],
*args: Any,
**kwargs: Any
) -> 'StringType':
if isinstance(source, (bytes, BytesType)):
return super().__new__(cls, source.decode('utf'))
elif isinstance(source, (str, StringType)):
# TODO: Consider returning the original StringType object.
return super().__new__(cls, source)
else:
return cast(StringType, super().__new__(cls, source))
def __repr__(self) -> str:
return f"{self.__class__.__name__}({super().__repr__()})"
@type_matched
def __eq__(self, other: Any) -> bool:
return super().__eq__(other)
@type_matched
def __ne__(self, other: Any) -> bool:
return super().__ne__(other)
def __hash__(self) -> int:
return super().__hash__()
class TimestampType(datetime.datetime):
"""
Implements google.protobuf.Timestamp
See https://developers.google.com/protocol-buffers/docs/reference/google.protobuf
Also see https://www.ietf.org/rfc/rfc3339.txt.
The protobuf implementation is an ordered pair of int64 seconds and int32 nanos.
Instead of a Tuple[int, int] we use a wrapper for :py:class:`datetime.datetime`.
From protobuf documentation for making a Timestamp in Python::
now = time.time()
seconds = int(now)
nanos = int((now - seconds) * 10**9)
timestamp = Timestamp(seconds=seconds, nanos=nanos)
Also::
>>> t = TimestampType("2009-02-13T23:31:30Z")
>>> repr(t)
"TimestampType('2009-02-13T23:31:30Z')"
>>> t.timestamp()
1234567890.0
>>> str(t)
'2009-02-13T23:31:30Z'
:strong:`Timezones`
Timezones are expressed in the following grammar:
::
TimeZone = "UTC" | LongTZ | FixedTZ ;
LongTZ = ? list available at
http://joda-time.sourceforge.net/timezones.html ? ;
FixedTZ = ( "+" | "-" ) Digit Digit ":" Digit Digit ;
Digit = "0" | "1" | ... | "9" ;
Fixed timezones are explicit hour and minute offsets from UTC.
Long timezone names are like Europe/Paris, CET, or US/Central.
The Joda project (https://www.joda.org/joda-time/timezones.html)
says "Time zone data is provided by the public IANA time zone database."
The ``dateutil`` project (https://pypi.org/project/python-dateutil/)
is used for TZ handling and timestamp parsing.
Additionally, there is a ``TZ_ALIASES`` mapping available in this class to permit additional
timezone names. By default, the mapping is empty, and the only names
available are those recognized by :mod:`dateutil.tz`.
"""
TZ_ALIASES: Dict[str, str] = {}
def __new__(
cls: Type['TimestampType'],
source: Union[int, str, datetime.datetime],
*args: Any,
**kwargs: Any) -> 'TimestampType':
if isinstance(source, datetime.datetime):
# Wrap a datetime.datetime
return super().__new__(
cls,
year=source.year,
month=source.month,
day=source.day,
hour=source.hour,
minute=source.minute,
second=source.second,
microsecond=source.microsecond,
tzinfo=source.tzinfo or datetime.timezone.utc
)
elif isinstance(source, int) and len(args) >= 2:
# Wrap a sequence of integers that datetime.datetime might accept.
ts: TimestampType = super().__new__(
cls, source, *args, **kwargs
)
if not ts.tzinfo:
ts = ts.replace(tzinfo=datetime.timezone.utc)
return ts
elif isinstance(source, str):
# Use dateutil to try a variety of text formats.
parsed_datetime = dateutil.parser.isoparse(source)
return super().__new__(
cls,
year=parsed_datetime.year,
month=parsed_datetime.month,
day=parsed_datetime.day,
hour=parsed_datetime.hour,
minute=parsed_datetime.minute,
second=parsed_datetime.second,
microsecond=parsed_datetime.microsecond,
tzinfo=parsed_datetime.tzinfo
)
else:
raise TypeError(f"Cannot create {cls} from {source!r}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}({str(self)!r})"
def __str__(self) -> str:
return self.strftime("%Y-%m-%dT%H:%M:%SZ")
def __add__(self, other: Any) -> 'TimestampType':
"""Timestamp + Duration -> Timestamp"""
result = super().__add__(other)
if result == NotImplemented:
return NotImplemented
return TimestampType(result)
def __radd__(self, other: Any) -> 'TimestampType':
"""Duration + Timestamp -> Timestamp"""
result = super().__radd__(other)
if result == NotImplemented:
return NotImplemented
return TimestampType(result)
# For more information, check the typeshed definition
# https://github.com/python/typeshed/blob/master/stdlib/2and3/datetime.pyi
@overload # type: ignore
def __sub__(self, other: 'TimestampType') -> 'DurationType':
... # pragma: no cover
@overload
def __sub__(self, other: 'DurationType') -> 'TimestampType':
... # pragma: no cover
def __sub__(
self,
other: Union['TimestampType', 'DurationType']
) -> Union['TimestampType', 'DurationType']:
result = super().__sub__(other)
if result == NotImplemented:
return cast(DurationType, result)
if isinstance(result, datetime.timedelta):
return DurationType(result)
return TimestampType(result)
@classmethod
def tz_name_lookup(cls, tz_name: str) -> Optional[datetime.tzinfo]:
"""
The :py:func:`dateutil.tz.gettz` may be extended with additional aliases.
.. TODO: Permit an extension into the timezone lookup.
Tweak ``celpy.celtypes.TimestampType.TZ_ALIASES``.
"""
tz_lookup = str(tz_name)
if tz_lookup in cls.TZ_ALIASES:
tz = dateutil.tz.gettz(cls.TZ_ALIASES[tz_lookup])
else:
tz = dateutil.tz.gettz(tz_lookup)
return tz
@classmethod
def tz_offset_parse(cls, tz_name: str) -> Optional[datetime.tzinfo]:
tz_pat = re.compile(r"^([+-]?)(\d\d?):(\d\d)$")
tz_match = tz_pat.match(tz_name)
if not tz_match:
raise ValueError(f"Unparsable timezone: {tz_name!r}")
sign, hh, mm = tz_match.groups()
offset_min = (int(hh) * 60 + int(mm)) * (-1 if sign == '-' else +1)
offset = datetime.timedelta(seconds=offset_min * 60)
tz = datetime.timezone(offset)
return tz
@staticmethod
def tz_parse(tz_name: Optional[str]) -> Optional[datetime.tzinfo]:
if tz_name:
tz = TimestampType.tz_name_lookup(tz_name)
if tz is None:
tz = TimestampType.tz_offset_parse(tz_name)
return tz
else:
return dateutil.tz.UTC
def getDate(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).day)
def getDayOfMonth(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).day - 1)
def getDayOfWeek(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).isoweekday() % 7)
def getDayOfYear(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
working_date = self.astimezone(new_tz)
jan1 = datetime.datetime(working_date.year, 1, 1, tzinfo=new_tz)
days = working_date.toordinal() - jan1.toordinal()
return IntType(days)
def getMonth(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).month - 1)
def getFullYear(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).year)
def getHours(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).hour)
def getMilliseconds(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).microsecond // 1000)
def getMinutes(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).minute)
def getSeconds(self, tz_name: Optional[StringType] = None) -> IntType:
new_tz = self.tz_parse(tz_name)
return IntType(self.astimezone(new_tz).second)
class DurationType(datetime.timedelta):
"""
Implements google.protobuf.Duration
https://developers.google.com/protocol-buffers/docs/reference/google.protobuf#duration
The protobuf implementation is an ordered pair of int64 seconds and int32 nanos.
Instead of a Tuple[int, int] we use a wrapper for :py:class:`datetime.timedelta`.
The definition once said this::
"type conversion, duration should be end with "s", which stands for seconds"
This is obsolete, however, considering the following issue.
See https://github.com/google/cel-spec/issues/138
This refers to the following implementation detail
::
// A duration string is a possibly signed sequence of
// decimal numbers, each with optional fraction and a unit suffix,
// such as "300ms", "-1.5h" or "2h45m".
// Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
The real regex, then is this::
[-+]?([0-9]*(\\.[0-9]*)?[a-z]+)+
"""
MaxSeconds = 315576000000
MinSeconds = -315576000000
NanosecondsPerSecond = 1000000000
scale: Dict[str, float] = {
"ns": 1E-9,
"us": 1E-6,
"µs": 1E-6,
"ms": 1E-3,
"s": 1.,
"m": 60.,
"h": 60. * 60.,
"d": 24. * 60. * 60.,
}
def __new__(
cls: Type['DurationType'],
seconds: Any,
nanos: int = 0,
**kwargs: Any) -> 'DurationType':
if isinstance(seconds, datetime.timedelta):
if not (cls.MinSeconds <= seconds.total_seconds() <= cls.MaxSeconds):
raise ValueError("range error: {seconds}")
return super().__new__(
cls, days=seconds.days, seconds=seconds.seconds, microseconds=seconds.microseconds)
elif isinstance(seconds, int):
if not (cls.MinSeconds <= seconds <= cls.MaxSeconds):
raise ValueError("range error: {seconds}")
return super().__new__(
cls, seconds=seconds, microseconds=nanos // 1000)
elif isinstance(seconds, str):
duration_pat = re.compile(r"^[-+]?([0-9]*(\.[0-9]*)?[a-z]+)+$")
duration_match = duration_pat.match(seconds)
if not duration_match:
raise ValueError(f"Invalid duration {seconds!r}")
# Consume the sign.
sign: float
if seconds.startswith("+"):
seconds = seconds[1:]
sign = +1
elif seconds.startswith("-"):
seconds = seconds[1:]
sign = -1
else:
sign = +1
# Sum the remaining time components: number * unit
try:
seconds = sign * fsum(
map(
lambda n_u: float(n_u.group(1)) * cls.scale[n_u.group(3)],
re.finditer(r"([0-9]*(\.[0-9]*)?)([a-z]+)", seconds)
)
)
except KeyError:
raise ValueError(f"Invalid duration {seconds!r}")
if not (cls.MinSeconds <= seconds <= cls.MaxSeconds):
raise ValueError("range error: {seconds}")
return super().__new__(
cls, seconds=seconds)
else:
raise TypeError(f"Invalid initial value type: {type(seconds)}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}({str(self)!r})"
def __str__(self) -> str:
return "{0}s".format(int(self.total_seconds()))
def __add__(self, other: Any) -> 'DurationType':
"""
This doesn't need to handle the rich variety of TimestampType overloadds.
This class only needs to handle results of duration + duration.
A duration + timestamp is not implemented by the timedelta superclass;
it is handled by the datetime superclass that implementes timestamp + duration.
"""
result = super().__add__(other)
if result == NotImplemented:
return cast(DurationType, result)
# This is handled by TimestampType; this is here for completeness, but isn't used.
if isinstance(result, (datetime.datetime, TimestampType)):
return TimestampType(result) # pragma: no cover
return DurationType(result)
def __radd__(self, other: Any) -> 'DurationType': # pragma: no cover
"""
This doesn't need to handle the rich variety of TimestampType overloadds.
Most cases are handled by TimeStamp.
"""
result = super().__radd__(other)
if result == NotImplemented:
return cast(DurationType, result)
# This is handled by TimestampType; this is here for completeness, but isn't used.
if isinstance(result, (datetime.datetime, TimestampType)):
return TimestampType(result)
return DurationType(result)
def getHours(self, tz_name: Optional[str] = None) -> IntType:
assert tz_name is None
return IntType(int(self.total_seconds() / 60 / 60))
def getMilliseconds(self, tz_name: Optional[str] = None) -> IntType:
assert tz_name is None
return IntType(int(self.total_seconds() * 1000))
def getMinutes(self, tz_name: Optional[str] = None) -> IntType:
assert tz_name is None
return IntType(int(self.total_seconds() / 60))
def getSeconds(self, tz_name: Optional[str] = None) -> IntType:
assert tz_name is None
return IntType(int(self.total_seconds()))
class FunctionType:
"""
We need a concrete Annotation object to describe callables to celpy.
We need to describe functions as well as callable objects.
The description would tend to shadow ``typing.Callable``.
An ``__isinstance__()`` method, for example, may be helpful for run-time type-checking.
Superclass for CEL extension functions that are defined at run-time.
This permits a formal annotation in the environment construction that creates
an intended type for a given name.
This allows for some run-time type checking to see if the actual object binding
matches the declared type binding.
Also used to define protobuf classes provided as an annotation.
We *could* define this as three overloads to cover unary, binary, and tertiary cases.
"""
def __call__(self, *args: Value, **kwargs: Value) -> Value:
raise NotImplementedError
class PackageType(MapType):
"""
A package of message types, usually protobuf.
TODO: This may not be needed.
"""
pass
class MessageType(MapType):
"""
An individual protobuf message definition. A mapping from field name to field value.
See Scenario: "message_literal" in the parse.feature. This is a very deeply-nested
message (30? levels), but the navigation to "payload" field seems to create a default
value at the top level.
"""
def __init__(self, *args: Value, **fields: Value) -> None:
if args and len(args) == 1:
super().__init__(cast(Mapping[Value, Value], args[0]))
elif args and len(args) > 1:
raise TypeError(r"Expected dictionary or fields, not {args!r}")
else:
super().__init__({StringType(k): v for k, v in fields.items()})
# def get(self, field: Any, default: Optional[Value] = None) -> Value:
# """
# Alternative implementation with descent to locate a deeply-buried field.
# It seemed like this was the defined behavior. It turns it, it isn't.
# The code is here in case we're wrong and it really is the defined behavior.
#
# Note. There is no default provision in CEL.
# """
# if field in self:
# return super().get(field)
#
# def descend(message: MessageType, field: Value) -> MessageType:
# if field in message:
# return message
# for k in message.keys():
# found = descend(message[k], field)
# if found is not None:
# return found
# return None
#
# sub_message = descend(self, field)
# if sub_message is None:
# return default
# return sub_message.get(field)
class TypeType:
"""
Annotation used to mark protobuf type objects.
We map these to CELTypes so that type name testing works.
"""
type_name_mapping = {
"google.protobuf.Duration": DurationType,
"google.protobuf.Timestamp": TimestampType,
"google.protobuf.Int32Value": IntType,
"google.protobuf.Int64Value": IntType,
"google.protobuf.UInt32Value": UintType,
"google.protobuf.UInt64Value": UintType,
"google.protobuf.FloatValue": DoubleType,
"google.protobuf.DoubleValue": DoubleType,
"google.protobuf.Value": MessageType,
"google.protubuf.Any": MessageType, # Weird.
"google.protobuf.Any": MessageType,
"list_type": ListType,
"map_type": MapType,
"map": MapType,
"list": ListType,
"string": StringType,
"bytes": BytesType,
"bool": BoolType,
"int": IntType,
"uint": UintType,
"double": DoubleType,
"null_type": type(None),
"STRING": StringType,
"BOOL": BoolType,
"INT64": IntType,
"UINT64": UintType,
"INT32": IntType,
"UINT32": UintType,
"BYTES": BytesType,
"DOUBLE": DoubleType,
}
def __init__(
self,
value: Any = "") -> None:
if isinstance(value, str) and value in self.type_name_mapping:
self.type_reference = self.type_name_mapping[value]
elif isinstance(value, str):
try:
self.type_reference = eval(value)
except (NameError, SyntaxError):
raise TypeError(f"Unknown type {value!r}")
else:
self.type_reference = value.__class__
def __eq__(self, other: Any) -> bool:
return (
other == self.type_reference
or isinstance(other, self.type_reference) # noqa: W503
)
