pynenc.trigger.arguments.argument_filters

Argument filters for trigger-based task executions.

This module defines abstractions and implementations for filtering task arguments based on various conditions. It enables flexible matching of task arguments against filter criteria.

Module Contents

Classes

ArgumentFilter

Base class for argument filters.

StaticArgumentFilter

Filters task arguments based on exact matching with provided arguments.

CallableArgumentFilter

Filters task arguments using a custom callable.

Functions

_compute_arguments_hash

Compute deterministic hash from arguments dictionary.

create_argument_filter

Factory function to create the appropriate argument filter based on the input type.

API

pynenc.trigger.arguments.argument_filters._compute_arguments_hash(arguments: dict[str, Any]) → str[source]

Compute deterministic hash from arguments dictionary.

Sorts arguments recursively to ensure consistent hashing regardless of insertion order. Uses JSON for serialization of standard types.

Parameters:

arguments (dict[str, Any]) – Arguments to hash

Returns:

SHA256 hash of the arguments

class pynenc.trigger.arguments.argument_filters.ArgumentFilter[source]

Bases: abc.ABC

Base class for argument filters.

Argument filters determine if task arguments match certain filter criteria.

_argument_filter_class_cache: ClassVar[dict[str, type[pynenc.trigger.arguments.argument_filters.ArgumentFilter]]]

None

_cache_initialized: ClassVar[bool]

False

classmethod _initialize_class_cache() → None[source]

Initialize the argument filter class cache by recursively finding all subclasses.

classmethod get_argument_filter_class(argument_filter_type: str) → type[pynenc.trigger.arguments.argument_filters.ArgumentFilter] | None[source]

Get an argument filter class by its name from the class cache.

Parameters:

argument_filter_type – Name of the argument filter class to find

Returns:

The argument filter class or None if not found

abstractmethod filter_id() → str

Unique identifier for this argument filter.

This identifier is used to distinguish between different argument filter types.

abstractmethod filter_arguments(arguments: dict[str, Any]) → bool[source]

Check if the given arguments match the filter criteria.

Parameters:

arguments – Task arguments to check

Returns:

True if arguments match the filter, False otherwise

to_json(app: pynenc.app.Pynenc) → str[source]

Serialize this argument filter to a JSON string.

Parameters:

app – Pynenc application instance for serializing complex arguments

Returns:

JSON string representation of this argument filter

abstractmethod _to_json(app: pynenc.app.Pynenc) → dict[str, Any][source]

Create a serializable representation of this argument filter.

Subclasses must implement this method to handle their specific serialization logic.

Parameters:

app – Pynenc application instance for serializing complex arguments

Returns:

Dictionary with serialized argument filter data

classmethod from_json(json_str: str, app: pynenc.app.Pynenc) → pynenc.trigger.arguments.argument_filters.ArgumentFilter[source]

Create an argument filter instance from a JSON string.

This is a factory method that instantiates the correct subclass based on the filter_type field in the JSON data.

Parameters:

  • json_str – JSON string containing serialized argument filter data

  • app – Pynenc application instance for deserializing complex arguments

Returns:

A new instance of the appropriate ArgumentFilter subclass

Raises:

ValueError – If the JSON data is invalid or the filter type is unknown

abstractmethod classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.arguments.argument_filters.ArgumentFilter[source]

Create an argument filter instance from parsed JSON data.

Each subclass must implement this method to handle its specific deserialization logic.

Parameters:

  • data – Dictionary with argument filter data from JSON

  • app – Pynenc application instance for deserializing complex arguments

Returns:

A new instance of this argument filter class

Raises:

ValueError – If the data is invalid for this argument filter type

class pynenc.trigger.arguments.argument_filters.StaticArgumentFilter(arguments: dict[str, Any])[source]

Bases: pynenc.trigger.arguments.argument_filters.ArgumentFilter

Filters task arguments based on exact matching with provided arguments.

Stores filter arguments as a plain dictionary for direct comparison.

Initialization

Initialize with filter arguments.

Parameters:

arguments – Arguments that task arguments must match

filter_id() → str

Generate a unique ID for this argument filter.

The ID is based on a deterministic hash of the filter arguments, handling both JSON-serializable and complex types.

filter_arguments(arguments: dict[str, Any]) → bool[source]

Check if the given arguments exactly match the filter arguments.

Parameters:

arguments – Task arguments to check

Returns:

True if arguments match the filter, False otherwise

_to_json(app: pynenc.app.Pynenc) → dict[str, Any][source]

Create a serializable representation of this argument filter.

Parameters:

app – Pynenc application instance for serializing complex arguments

Returns:

Dictionary with serialized argument filter data

classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.arguments.argument_filters.StaticArgumentFilter[source]

Create an StaticArgumentFilter instance from parsed JSON data.

Parameters:

  • data – Dictionary with argument filter data from JSON

  • app – Pynenc application instance for deserializing complex arguments

Returns:

A new StaticArgumentFilter instance

__eq__(other: Any) → bool[source]

Check equality with another object.

Parameters:

other – Object to compare with

Returns:

True if equal, False otherwise

class pynenc.trigger.arguments.argument_filters.CallableArgumentFilter(callable_filter: collections.abc.Callable[[dict[str, Any]], bool] | pynenc.trigger.arguments.arguments_common.SerializableCallable)[source]

Bases: pynenc.trigger.arguments.argument_filters.ArgumentFilter

Filters task arguments using a custom callable.

This provides maximum flexibility for complex filtering logic.

Initialization

Initialize with a callable filter.

Parameters:

callable_filter – Function that takes arguments and returns a boolean

filter_id() → str

Generate a unique ID for this argument filter.

The ID is based on the ID of the callable filter.

filter_arguments(arguments: dict[str, Any]) → bool[source]

Apply the callable filter to the arguments.

Parameters:

arguments – Task arguments to check

Returns:

True if arguments match the filter, False otherwise

_to_json(app: pynenc.app.Pynenc) → dict[str, Any][source]

Create a serializable representation of this argument filter.

Parameters:

app – Pynenc application instance for serializing complex arguments

Returns:

Dictionary with serialized argument filter data

classmethod _from_json(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.arguments.argument_filters.CallableArgumentFilter[source]

Create a CallableArgumentFilter instance from parsed JSON data.

Parameters:

  • data – Dictionary with argument filter data from JSON

  • app – Pynenc application instance for deserializing complex arguments

Returns:

A new CallableArgumentFilter instance

pynenc.trigger.arguments.argument_filters.create_argument_filter(filter_spec: None | dict[str, Any] | collections.abc.Callable[[dict[str, Any]], bool]) → pynenc.trigger.arguments.argument_filters.ArgumentFilter[source]

Factory function to create the appropriate argument filter based on the input type.

Parameters:

filter_spec – Either a dictionary of arguments or a callable function

Returns:

An instance of ArgumentFilter