#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# rule_engine/builtins.py
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of the project nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
import collections
import collections.abc
import datetime
import decimal
import functools
import math
import random
from typing import Any, Callable, Iterable, Iterator, Mapping
from . import errors
from . import types
from .parser.utilities import parse_datetime, parse_float, parse_timedelta
import dateutil.tz
def _builtin_filter(function: Callable[[Any], Any], iterable: Iterable[Any]) -> tuple[Any, ...]:
return tuple(filter(function, iterable))
def _builtin_map(function: Callable[[Any], Any], iterable: Iterable[Any]) -> tuple[Any, ...]:
return tuple(map(function, iterable))
def _builtin_parse_datetime(builtins: 'Builtins', string: str) -> datetime.datetime:
return parse_datetime(string, builtins.timezone)
def _builtin_random(boundary: Any = None) -> Any:
if boundary:
if not types.is_natural_number(boundary):
raise errors.FunctionCallError('argument #1 (boundary) must be a natural number')
return random.randint(0, int(boundary))
return random.random()
def _builtin_now(builtins: 'Builtins') -> datetime.datetime:
return datetime.datetime.now(tz=builtins.timezone)
def _builtin_today(builtins: 'Builtins') -> datetime.datetime:
return _builtin_now(builtins).replace(hour=0, minute=0, second=0, microsecond=0)
def _builtin_parse_datetime_generator(builtins: 'Builtins') -> 'functools.partial[datetime.datetime]':
return functools.partial(_builtin_parse_datetime, builtins)
def _builtin_range(start: Any, stop: Any = None, step: Any = None) -> list[int]:
if not types.is_integer_number(start):
raise errors.FunctionCallError('argument #1 (start) must be an integer number')
if stop:
if not types.is_integer_number(stop):
raise errors.FunctionCallError('argument #2 (stop) must be an integer number')
if step:
if not types.is_integer_number(step):
raise errors.FunctionCallError('argument #3 (step) must be an integer number')
return list(range(int(start), int(stop), int(step)))
return list(range(int(start), int(stop)))
return list(range(int(start)))
def _builtins_split(string: str, sep: str | None = None, maxsplit: Any = None) -> tuple[str, ...]:
if maxsplit is None:
maxsplit = -1
elif types.is_natural_number(maxsplit):
maxsplit = int(maxsplit)
else:
raise errors.FunctionCallError('argument #3 (maxsplit) must be a natural number')
return tuple(string.split(sep=sep, maxsplit=maxsplit))
class BuiltinValueGenerator(object):
"""
A class used as a wrapper for builtin values to differentiate between a value that is a function and a value that
should be generated by calling a function. A value that is generated by calling a function is useful for determining
the value during evaluation for things like the current time.
.. versionadded:: 4.0.0
"""
__slots__ = ('callable',)
callable: Callable[['Builtins'], Any]
def __init__(self, callable: Callable[['Builtins'], Any]) -> None:
self.callable = callable
def __call__(self, builtins: 'Builtins') -> Any:
return self.callable(builtins)
[docs]
class Builtins(collections.abc.Mapping):
"""
A class to define and provide variables to within the builtin context of rules. These can be accessed by specifying
a symbol name with the ``$`` prefix.
"""
scope_name = 'built-in'
"""The identity name of the scope for builtin symbols."""
[docs]
def __init__(
self,
values: Mapping[str, Any],
namespace: str | None = None,
timezone: datetime.tzinfo | None = None,
value_types: Mapping[str, 'types._DataTypeDef'] | None = None
) -> None:
"""
:param dict values: A mapping of string keys to be used as symbol names with values of either Python literals or
a function which will be called when the symbol is accessed. When using a function, it will be passed a
single argument, which is the instance of :py:class:`Builtins`.
:param str namespace: The namespace of the variables to resolve.
:param timezone: A timezone to use when resolving timestamps.
:type timezone: :py:class:`~datetime.tzinfo`
:param dict value_types: A mapping of the values to their datatypes.
.. versionchanged:: 2.3.0
Added the *value_types* parameter.
"""
self.__values = values
self.__value_types = value_types or {}
self.namespace = namespace
self.timezone = timezone or dateutil.tz.tzlocal()
[docs]
def resolve_type(self, name: str) -> 'types._DataTypeDef':
"""
The method to use for resolving the data type of a builtin symbol.
:param str name: The name of the symbol to retrieve the data type of.
:return: The data type of the symbol or :py:attr:`~rule_engine.types.DataType.UNDEFINED`.
"""
return self.__value_types.get(name, types.DataType.UNDEFINED)
def __repr__(self) -> str:
return "<{} namespace={!r} keys={!r} timezone={!r} >".format(self.__class__.__name__, self.namespace, tuple(self.keys()), self.timezone)
def __getitem__(self, name: str) -> Any:
value = self.__values[name]
if isinstance(value, collections.abc.Mapping):
if self.namespace is None:
namespace = name
else:
namespace = self.namespace + '.' + name
return self.__class__(value, namespace=namespace, timezone=self.timezone)
elif callable(value) and isinstance(value, BuiltinValueGenerator):
value = value(self)
return value
def __iter__(self) -> Iterator[str]:
return iter(self.__values)
def __len__(self) -> int:
return len(self.__values)
[docs]
@classmethod
def from_defaults(cls, values: Mapping[str, Any] | None = None, **kwargs: Any) -> 'Builtins':
"""Initialize a :py:class:`Builtins` instance with a set of default values."""
now = BuiltinValueGenerator(_builtin_now)
# there may be errors here if the decimal.Context precision exceeds what is provided by the math constants
default_values = {
# mathematical constants
'e': decimal.Decimal(repr(math.e)),
'pi': decimal.Decimal(repr(math.pi)),
# timestamps
'now': now,
'today': BuiltinValueGenerator(_builtin_today),
# functions
'abs': abs,
'any': any,
'all': all,
'sum': sum,
'map': _builtin_map,
'max': max,
'min': min,
'filter': _builtin_filter,
'parse_datetime': BuiltinValueGenerator(_builtin_parse_datetime_generator),
'parse_float': parse_float,
'parse_timedelta': parse_timedelta,
'random': _builtin_random,
'range': _builtin_range,
'split': _builtins_split
}
default_values.update(values or {})
default_value_types = {
# mathematical constants
'e': types.DataType.FLOAT,
'pi': types.DataType.FLOAT,
# timestamps
'now': types.DataType.DATETIME,
'today': types.DataType.DATETIME,
# functions
'abs': types.DataType.FUNCTION('abs', return_type=types.DataType.FLOAT, argument_types=(types.DataType.FLOAT,)),
'all': types.DataType.FUNCTION('all', return_type=types.DataType.BOOLEAN, argument_types=(types.DataType.ARRAY,)),
'any': types.DataType.FUNCTION('any', return_type=types.DataType.BOOLEAN, argument_types=(types.DataType.ARRAY,)),
'sum': types.DataType.FUNCTION('sum', return_type=types.DataType.FLOAT, argument_types=(types.DataType.ARRAY(types.DataType.FLOAT),)),
'map': types.DataType.FUNCTION('map', return_type=types.DataType.ARRAY, argument_types=(types.DataType.FUNCTION, types.DataType.ARRAY)),
'max': types.DataType.FUNCTION('max', return_type=types.DataType.FLOAT, argument_types=(types.DataType.ARRAY(types.DataType.FLOAT),)),
'min': types.DataType.FUNCTION('min', return_type=types.DataType.FLOAT, argument_types=(types.DataType.ARRAY(types.DataType.FLOAT),)),
'filter': types.DataType.FUNCTION('filter', return_type=types.DataType.ARRAY, argument_types=(types.DataType.FUNCTION, types.DataType.ARRAY)),
'parse_datetime': types.DataType.FUNCTION('parse_datetime', return_type=types.DataType.DATETIME, argument_types=(types.DataType.STRING,)),
'parse_float': types.DataType.FUNCTION('parse_float', return_type=types.DataType.FLOAT, argument_types=(types.DataType.STRING,)),
'parse_timedelta': types.DataType.FUNCTION('parse_timedelta', return_type=types.DataType.TIMEDELTA, argument_types=(types.DataType.STRING,)),
'random': types.DataType.FUNCTION('random', return_type=types.DataType.FLOAT, argument_types=(types.DataType.FLOAT,), minimum_arguments=0),
'range': types.DataType.FUNCTION('range', return_type=types.DataType.ARRAY(types.DataType.FLOAT), argument_types=(types.DataType.FLOAT, types.DataType.FLOAT, types.DataType.FLOAT,), minimum_arguments=1),
'split': types.DataType.FUNCTION(
'split',
return_type=types.DataType.ARRAY(types.DataType.STRING),
argument_types=(types.DataType.STRING, types.DataType.STRING, types.DataType.FLOAT),
minimum_arguments=1
)
}
default_value_types.update(kwargs.pop('value_types', {}))
return cls(default_values, value_types=default_value_types, **kwargs)