engine

This module contains the primary externally facing API for the package.

Functions

resolve_attribute(thing: Any, name: str) → Any[source]

A replacement resolver function for looking up symbols as members of thing. This is effectively the same as thing.name. The thing object can be a namedtuple(), a custom Python class or any other object. Each of the members of thing must be of a compatible data type.

Warning

This effectively exposes all members of thing. If any members are sensitive, then a custom resolver should be used that checks name against a whitelist of attributes that are allowed to be accessed.

Parameters:

  • thing – The object on which the name attribute will be accessed.

  • name (str) – The symbol name that is being resolved.

Returns:

The value for the corresponding attribute name.

resolve_item(thing: Any, name: str) → Any[source]

A resolver function for looking up symbols as items from an object (thing) which supports the Mapping interface, such as a dictionary. This is effectively the same as thing['name']. Each of the values in thing must be of a compatible data type.

Parameters:

  • thing – The object from which the name item will be accessed.

  • name (str) – The symbol name that is being resolved.

Returns:

The value for the corresponding attribute name.

type_resolver_from_dataclass(cls: type, *, strict: bool = True) → Callable[[str], _DataTypeDef][source]

Return a function suitable for use as the type_resolver for a Context instance from a Python dataclass(). The dataclass’s top-level fields become resolvable symbols (so a rule may reference them by name) and every transitively-reachable OBJECT schema is registered by its OBJECT name so that cross-type references in mutually-recursive schemas resolve at parse time.

The companion from_dataclass() is used internally to derive the schema; see its documentation for the supported annotation forms (primitives, generics, Optional, nested dataclasses and self/mutual recursion). Nullable fields (annotated Optional[T] or T | None) surface as NULLABLE wrappers in the resolved symbol types.

Added in version 5.0.0.

Parameters:
Returns:

The callback function.

Return type:

function

type_resolver_from_dict(dictionary: Mapping[str, Any]) → Callable[[str], _DataTypeDef][source]

Return a function suitable for use as the type_resolver for a Context instance from a dictionary. If any of the values within the dictionary are not of a compatible data type, a TypeError will be raised. Additionally, the resulting function will raise a SymbolResolutionError if the symbol name does not exist within the dictionary.

Parameters:

dictionary (dict) – A dictionary (or any other object which supports the Mapping interface) from which to create the callback function.

Returns:

The callback function.

Return type:

function

type_resolver_from_sqlalchemy(cls: type, *, strict: bool = True) → Callable[[str], _DataTypeDef][source]

Return a function suitable for use as the type_resolver for a Context instance from a SQLAlchemy ORM mapped class. The class’s top-level columns and relationships become resolvable symbols and every transitively-reachable OBJECT schema is registered by its OBJECT name so that cross-type references in mutually-recursive relationship graphs resolve at parse time.

The companion from_sqlalchemy() is used internally to derive the schema; see its documentation for the column → DataType mapping and how relationships are expanded into nested OBJECT types. Columns with nullable=True and scalar relationships whose local foreign-key columns are nullable surface as NULLABLE wrappers in the resolved symbol types.

SQLAlchemy is an optional dependency; the import is deferred until this function is actually called.

Added in version 5.0.0.

Parameters:
Returns:

The callback function.

Return type:

function

Classes

class Chain(entries: Iterable[tuple[Rule | str, Any]] | None = None, *, default_value: Any = UNDEFINED, context: Context | None = None)[source]

Bases: object

An ordered collection of rules, each of which is paired with a value. An object is evaluated against each rule in order and the value of the first rule that matches it is returned, much like a packet traversing a chain in iptables where the value is the target (e.g. ACCEPT or DROP) and the default value is the chain’s policy.

A rule matches when Rule.matches() returns a truthy value. Exceptions raised while evaluating a rule, such as a SymbolResolutionError, are not caught and stop the evaluation.

Evaluating a chain does not acquire a lock. Each evaluation uses the entries that were present when it started, so one thread can modify a chain while others are evaluating it. Modifications are serialized so concurrent changes are not lost, however adding or removing an entry copies the others, so a large chain is best built by passing its entries to the constructor. A chain should not be modified from a __del__ method or a weakref finalizer because the garbage collector can run them while a modification is in progress, which would deadlock. A chain can be pickled as long as its context, rules and values can be.

Indexes start at 0, unlike the rule numbers used by iptables, and negative indexes count from the end of the chain. Slicing a chain returns a new chain with the entries in the slice and the same default_value and context. Adding or removing entries in either chain afterwards does not change the other.

Added in version 5.1.0.

__init__(entries: Iterable[tuple[Rule | str, Any]] | None = None, *, default_value: Any = UNDEFINED, context: Context | None = None) → None[source]
Parameters:

  • entries – An optional iterable of (rule, value) pairs to initialize the chain with, in order. Each rule can be either a Rule instance or the text of one, which will be compiled using context.

  • default_value – The value that evaluate() returns when no rule matches. If this is UNDEFINED (the default), a ChainExhaustedError is raised instead.

  • context (Context) – The context to use for compiling rules that are specified as text. If not specified, a new default Context is created and shared by all of them. Rule instances are used as-is and keep their own context.

append(rule: Rule | str, value: Any) → ChainEntry[source]

Add a new entry to the end of the chain, like iptables --append.

Parameters:

  • rule – The rule for the new entry, either a Rule instance or the text of one, which will be compiled using context.

  • value – The value to return when rule matches.

Returns:

The new entry.

Return type:

ChainEntry

context

The context that is used for compiling rules that are specified as text.

default_value

The default_value parameter from __init__().

delete(index: int) → ChainEntry[source]

Remove the entry at index from the chain, like iptables --delete. Indexes start at 0 and negative indexes count from the end of the chain.

Parameters:

index (int) – The index of the entry to remove.

Returns:

The entry that was removed.

Return type:

ChainEntry

Raises:

IndexError – There is no entry at index.

evaluate(thing: Any) → Any[source]

Evaluate thing against each rule in the chain and return the value of the first one that matches. If no rule matches, the default_value is returned.

Parameters:

thing – The object on which to apply the rules.

Returns:

The value of the first matching rule, or the default value.

Raises:

ChainExhaustedError – No rule matched and default_value is UNDEFINED.

find(thing: Any) → ChainEntry | None[source]

Evaluate thing against each rule in the chain and return the first entry that matches. Unlike evaluate(), the default value is not used. This can be used to determine which rule matched, and index() can then be used to determine its position.

Parameters:

thing – The object on which to apply the rules.

Returns:

The first matching ChainEntry, or None if no rule matched.

flush() → None[source]

Remove all entries from the chain, like iptables --flush. The default_value is unchanged.

index(entry: ChainEntry) → int[source]

Return the index of entry in the chain, such as an entry that was returned by find(). Entries are compared as tuples and a rule is only equal to itself, so an entry is only found in a chain that contains the same rule, like the chain it came from or a slice of it.

Parameters:

entry (ChainEntry) – The entry to look up.

Returns:

The index of the first entry that is equal to entry.

Return type:

int

Raises:

ValueError – entry is not in the chain.

insert(index: int, rule: Rule | str, value: Any) → ChainEntry[source]

Insert a new entry into the chain before index, like iptables --insert. Indexes start at 0 and negative indexes count from the end of the chain. This follows the semantics of list.insert(), so an index past either end of the chain adds the entry to that end.

Parameters:

  • index (int) – The index to insert the new entry before.

  • rule – The rule for the new entry, either a Rule instance or the text of one, which will be compiled using context.

  • value – The value to return when rule matches.

Returns:

The new entry.

Return type:

ChainEntry

class ChainEntry(rule: Rule, value: Any)[source]

Bases: NamedTuple

A single entry within a Chain, pairing a rule with the value that is returned when it matches.

Added in version 5.1.0.

rule: Rule

The Rule that objects are evaluated against.

value: Any

The value that is returned by Chain.evaluate() when rule matches.

class Context(*, regex_flags: int = 0, resolver: Callable[[Any, str], Any] | None = None, type_resolver: Callable[[str], _DataTypeDef] | Mapping[str, Any] | None = None, default_timezone: str | tzinfo = 'local', default_value: Any = UNDEFINED, decimal_context: Context | None = None, mapping_attribute_lookup: bool = True, int_max_digits: int | None = None)[source]

Bases: object

An object defining the context for a rule’s evaluation. This can be used to change the behavior of certain aspects of the rule such as how symbols are resolved and what regex flags should be used.

__init__(*, regex_flags: int = 0, resolver: Callable[[Any, str], Any] | None = None, type_resolver: Callable[[str], _DataTypeDef] | Mapping[str, Any] | None = None, default_timezone: str | tzinfo = 'local', default_value: Any = UNDEFINED, decimal_context: Context | None = None, mapping_attribute_lookup: bool = True, int_max_digits: int | None = None) → None[source]
Parameters:

  • regex_flags (int) – The flags to provide to functions in the re module when calling either the match() or search() functions for comparison expressions.

  • resolver – An optional callback function to use in place of resolve().

  • type_resolver (function, dict) – An optional callback function to use in place of resolve_type().

  • default_timezone (str, tzinfo) – The default timezone to apply to datetime instances which do not have one specified. This is necessary for comparison operations. The value should either be a tzinfo instance, or a string. If default_timzezone is a string it must be one of the specially supported (case-insensitive) values of “local” or “utc”.

  • default_value – The default value to return when resolving either a missing symbol or attribute.

  • decimal_context – A specific decimal.Context object to use for evaluation of FLOAT values. The default value will be taken from the current thread and will be used by all evaluations using this Context regardless of the decimal context of the thread which evaluates the rule. This causes the rule evaluation to be consistent regardless of the calling thread.

  • mapping_attribute_lookup (bool) – Whether or not to allow attribute-style access on MAPPING values as a fallback to key-style lookup. When True (the default), accessing mapping.key falls back to mapping['key'] and emits a MappingAttributeLookupDeprecation once per context. When False, attribute access on a MAPPING raises a parse-time error. This fallback is scheduled for removal in v6.0.

  • int_max_digits (int) – The maximum number of digits that an integer can have to be converted to a FLOAT value, such as an integer resolved from a symbol or the result of a left shift. Converting an integer takes time that grows quadratically with its size, so one with more digits raises an ArithmeticError instead. 0 disables the limit and None (the default) uses the interpreter’s limit from sys.get_int_max_str_digits() each time a value is converted, including while a rule is being parsed. The limit applies wherever an integer is converted, including the results of functions and attributes such as floor and length, binary, octal and hexadecimal literals and integers in SET values, so a very small limit can reject ordinary values. Values that are already FLOAT values, such as the literal 1e25, are not checked.

Changed in version 2.0.0: Added the default_value parameter.

Changed in version 2.1.0: If type_resolver is a dictionary, type_resolver_from_dict() will be called on it automatically and the result will be used as the callback.

Changed in version 3.0.0: Added the decimal_context parameter.

Changed in version 5.0.0: Added the mapping_attribute_lookup parameter.

Changed in version 5.2.0: Added the int_max_digits parameter.

assignments(*assignments: Assignment) → Iterator[None][source]

Add the specified assignments to a thread-specific scope. This is used when an assignment originates from an expression.

Parameters:

assignments (Assignment) – The one or more assignments to define.

builtins

An instance of Builtins to provided a default set of builtin symbol values.

decimal_context

The decimal_context parameter from __init__()

default_timezone

The default_timezone parameter from __init__()

default_value

The default_value parameter from __init__()

int_max_digits

The int_max_digits parameter from __init__().

mapping_attribute_lookup

The mapping_attribute_lookup parameter from __init__().

regex_flags

The regex_flags parameter from __init__()

resolve(thing: Any, name: str, scope: str | None = None) → Any[source]

The method to use for resolving symbols names to values. This function must return a compatible value for the specified symbol name. When a scope is defined, this function handles the resolution itself, however when the scope is None the resolver specified in __init__() is used which defaults to resolve_item().

If name fails to resolve, this method will raise SymbolResolutionError. It is then up to the caller to determine whether or not it is appropriate to use default_value.

Parameters:

  • thing – The object from which the name item will be accessed.

  • name (str) – The symbol name that is being resolved.

Returns:

The value for the corresponding symbol name.

resolve_attribute(thing: Any, object_: Any, name: str) → Any[source]

The method to use for resolving attributes from values. This function must return a compatible value for the specified attribute name.

If name fails to resolve, this method will raise AttributeResolutionError. It is then up to the caller to determine whether or not it is appropriate to use default_value.

Parameters:

  • thing – The object from which the object_ was retrieved.

  • object – The object from which the name attribute will be accessed.

  • name (str) – The attribute name that is being resolved.

Returns:

The value for the corresponding attribute name.

resolve_attribute_type(object_type: _DataTypeDef, name: str) → _DataTypeDef

The method to use for resolving the data type of an attribute.

Parameters:

  • object_type – The data type of the object that name is an attribute of.

  • name (str) – The name of the attribute to retrieve the data type of.

Returns:

The data type of the specified attribute.

resolve_type(name: str, scope: str | None = None) → _DataTypeDef[source]

A method for providing type hints while the rule is being generated. This can be used to ensure that all symbol names are valid and that the types are appropriate for the operations being performed. It must then return one of the compatible data type constants if the symbol is valid or raise an exception. The default behavior is to return UNDEFINED for all symbols.

Parameters:

  • name (str) – The symbol name to provide a type hint for.

  • scope (str) – An optional scope name that identifies from where to resolve the name.

Returns:

The type of the specified symbol.

symbols: set[str]

The symbols that are referred to by the rule. Some or all of these will need to be resolved at evaluation time. This attribute can be used after a rule is generated to ensure that all symbols are valid before it is evaluated.

class Rule(text: str, context: Context | None = None)[source]

Bases: object

A rule which parses a string with a logical expression and can then evaluate an arbitrary object for whether or not it matches based on the constraints of the expression.

__init__(text: str, context: Context | None = None) → None[source]
Parameters:

  • text (str) – The text of the logical expression.

  • context (Context) – The context to use for evaluating the expression on arbitrary objects. This can be used to change the default behavior. The default context is Context but any object providing the same interface (such as a subclass) can be used.

evaluate(thing: Any) → Any[source]

Evaluate the rule against the specified thing and return the value. This can be used to, for example, apply the symbol resolver.

Parameters:

thing – The object on which to apply the rule.

Returns:

The value the rule evaluates to. Unlike the matches() method, this is not necessarily a boolean.

filter(things: Iterable[Any]) → Iterator[Any][source]

A convenience function for iterating over things and yielding each member that matches() return True for.

Parameters:

things – The collection of objects to iterate over.

classmethod is_valid(text: str, context: Context | None = None) → bool[source]

Test whether or not the rule is syntactically correct. This verifies the grammar is well structured and that there are no type compatibility issues regarding literals or symbols with known types (see resolve_type() for specifying symbol type information).

Parameters:

  • text (str) – The text of the logical expression.

  • context – The context as would be passed to the __init__() method. This can be used for specifying symbol type information.

Returns:

Whether or not the expression is well formed and appears valid.

Return type:

bool

matches(thing: Any) → bool[source]

Evaluate the rule against the specified thing. This will either return whether thing matches, or an exception will be raised.

Parameters:

thing – The object on which to apply the rule.

Returns:

Whether or not the rule matches.

Return type:

bool

parser: Parser = <rule_engine.parser.Parser object>

The Parser instance that will be used for parsing the rule text into a compatible abstract syntax tree (AST) for evaluation.

to_graphviz() → graphviz.Digraph[source]

Generate a diagram of the parsed rule’s AST using GraphViz.

Returns:

The rule diagram.

Return type:

graphviz.Digraph