Skip to content

Modules

predylogic.register

Classes

Registry

Bases: Generic[T_contra], Mapping[str, PredicateProducer[T_contra, ...]]

Registry a predicate producer with a name.

Functions
register(rule_def, alias)

Register a RuleDef with an optional alias. Maintain the original function unchanged.

Parameters:

Name Type Description Default
rule_def RuleDef[T_contra, P]

The rule definition to register.

required
alias str | None

An optional string representing the alias for the rule definition. When registering an anonymous function, this parameter must be provided.

required

Returns:

Type Description
PredicateProducer[T_contra, P]

PredicateProducer[P, T_contra]: A new predicate producer instance.

Raises:

Type Description
RegistryNameConflictError

If the name is already in use.

RuleDefNotNamedError

If a rule def is not named.

Examples:

Register a named rule definition:

from typing import TypedDict

from predylogic.register import Registry


class UserCtx(TypedDict):
    age: int

def is_user_over_age(user: User, age: int) -> bool:
    return user.age >= age

registry = Registry("my_registry")

# The original function remains unchanged.
is_user_over_age_rule = registry.register(is_user_over_age)
rule_def(alias=None)

Converts the given alias into a RuleDefConverter instance.

This method is used to create a RuleDefConverter object which can perform operations related to defining rules. The optional alias parameter can be supplied to distinguish or identify the rule definition.

Parameters:

Name Type Description Default
alias str | None

An optional string representing the alias for the rule definition.

None

Returns:

Type Description
RuleDefConverter[T_contra]

RuleDefConverter[T_contra]: A new instance of RuleDefConverter, configured

RuleDefConverter[T_contra]

with the provided alias.

Raises:

Type Description
RegistryNameConflictError

If the name is already in use.

RuleDefNotNamedError

If a rule def is not named.

Examples:

from typing import TypedDict

from predylogic import Registry, Predicate


class UserCtx(TypedDict):
    age: int


registry = Registry("my_registry")

@registry.rule_def()
def is_user_over_age(user: User, age: int) -> bool:
    return user.age >= age


is_legal = is_user_over_age(age=18)

assert isinstance(is_legal, Predicate)
assert is_user_over_age(18)

RegistryManager

Manage registries.

Functions
add_register(register)

Try to add a register.

Parameters:

Name Type Description Default
register Registry

The register to add.

required

Raises:

Type Description
RegisterNameConflictError

If the name is already in use.

get_register(name)

Get a register by name.

predylogic.predicate

Classes

ComposablePredicate

Bases: Predicate[T_contra], Protocol[T_contra]

Represents a predicate that can be composed using logical operations.

This class provides functionality for combining predicates using logical AND, OR, and NOT operations. It allows creating complex logical conditions by chaining or mixing these operations. It is particularly useful in scenarios requiring customizable condition evaluation, such as filtering or rule-based matching.

Functions
__and__(other)

Combine this predicate with another using logical AND.

To create a large number of consecutive and combinations, the all_of method should be used to avoid the overhead of creating additional objects.

__invert__()

Combine this predicate with another using logical not.

__or__(other)

Combine this predicate with another using logical or.

To create a large number of consecutive or combinations, the any_of method should be used to avoid the overhead of creating additional objects.

Predicate

Bases: Protocol[T_contra]

Defines a type-checkable protocol for predicates that can process a given context and return a boolean result or a trace of execution details when tracing is enabled.

This protocol is intended for use in scenarios where context-based decision logic is needed. It enforces the implementation of certain properties and method signatures, ensuring consistent behavior across different types of predicates.

Attributes:

Name Type Description
node_type PredicateNodeType

Specifies the type of the predicate node. It is mandatory for implementers of this protocol to provide this property.

desc str | None

An optional description of the predicate to provide additional context or information about its purpose.

name str | None

An optional name for the predicate, which can serve as an identifier or label.

Functions
__call__(ctx, /, *, trace=False, short_circuit=True, fail_skip=None)
__call__(
    ctx: T_contra,
    /,
    *,
    trace: Literal[True],
    short_circuit: bool = True,
    fail_skip: tuple[type[Exception], ...] | None = None,
) -> Trace
__call__(
    ctx: T_contra,
    /,
    *,
    trace: Literal[False] = False,
    short_circuit: Literal[True] = True,
    fail_skip: tuple[type[Exception], ...] | None = None,
) -> bool

Executes the callable object using the provided context and optional parameters to control execution behavior and error handling.

Parameters:

Name Type Description Default
ctx T_contra

The context that is passed into the callable object during execution.

required
trace bool

Specifies whether to enable tracing of execution for debugging or monitoring purposes. Defaults to False.

False
short_circuit bool

Determines whether the execution should stop immediately upon encountering a failure. Defaults to True.

True
fail_skip tuple[type[Exception], ...] | None

A tuple of exception types to be skipped or ignored during execution. If provided, these exceptions will not disrupt the execution's flow. Defaults to None.

None

Returns:

Type Description
bool | Trace

Either a boolean indicating the success or failure of the operation, or a

bool | Trace

Trace object that contains detailed execution history if tracing is enabled.

Functions

all_of(predicates)

Use this method to combine multiple predicates, thereby avoiding the object creation overhead associated with chained calls.

Parameters:

Name Type Description Default
predicates Sequence[ComposablePredicate[T_contra]]

Predicates are combined sequentially using and.

required

any_of(predicates)

Use this method to combine multiple predicates, thereby avoiding the object creation overhead associated with chained calls.

Parameters:

Name Type Description Default
predicates Sequence[ComposablePredicate[T_contra]]

Predicates are combined sequentially using or.

required

is_predicate(p)

Check if the given object is a valid predicate.

predylogic.rule_engine

Classes

RuleEngine

Handles the management, updating, and retrieval of rule set manifests and predicate handles.

The RuleEngine class is designed for managing rules and associated handlers. It provides functionalities to update rules and retrieve predicate handles for specified contexts. This is useful in systems where dynamic rule updates and context-specific predicate handling are required.

Attributes:

Name Type Description
registry_manager RegistryManager

Manages the registry of rule sets and their associated handlers.

Functions
get_predicate_handle(registry_name, rule_name, *, ctx_type=Any)
get_predicate_handle(
    registry_name: str, rule_name: str, *, ctx_type: type[T]
) -> PredicateHandle[T]
get_predicate_handle(
    registry_name: str, rule_name: str
) -> Predicate[Any]

Retrieves the handle for the specified predicate.

This method is used to get a handle for a predicate based on the provided name and context type.

The returned Predicate is a handle encapsulating the original Predicate. During update_manifests, atomic updates are performed on handles managed by the RuleEngine.

Parameters:

Name Type Description Default
registry_name str

name of the registry containing the predicate.

required
rule_name str

name of the predicate to retrieve.

required
ctx_type Any

The type of context associated with the predicate.

Any

Returns:

Type Description
Predicate[Any]

The handle for the specified predicate. If the predicate being looked up does not exist or has been deleted during a manifest update, a predicate will be returned that raises a RuleRevokedError exception.

update_manifests(*manifests)

Updates the provided manifests to the current instance.

This method is expected to take one or more manifests of type RuleSetManifest.

Rules bearing identical names within the same registry will be updated. Handles managed by the engine, where updated rules exist on the chain, will likewise be updated.

Parameters:

Name Type Description Default
manifests RuleSetManifest

One or more instances of RuleSetManifest that are to be updated.

()

SchemaGenerator

Bases: Generic[T_cap]

Generate JSON Schema for PredyLogic rules.

Attributes
rule_def_types cached property

Generates the union of rule definitions from the current registry.

This method creates a tuple of rule model definitions based on the current rule registry and combines them into a union type. If the registry is empty, it returns NoneType. The union definition includes a discriminator field for identifying specific rule definitions.

Returns:

Type Description
UnionType | type

A union of all the rule definitions created from the registry or

UnionType | type

NoneType if no definitions exist.

Functions
generate()

Generates a dynamic RuleSetManifest model with rebuilt validation.

The method dynamically constructs a RuleSetManifest model based on rule definitions fetched from the current object's context. The resulting model is renamed for clarity according to the current registry's name and its validation is rebuilt before returning the model.

Returns:

Type Description
type[RuleSetManifest]

A dynamically created and rebuilt RuleSetManifest

type[RuleSetManifest]

model specific to the registered rule definitions.