Labrea API

Core Types

class labrea.types.Evaluatable

Bases: Generic[A], Cacheable, Explainable, Validatable, ABC

Abstract base class for objects that can be evaluated with an Options dictionary.

This ABC is used to define a common interface for objects that can be evaluated using an options dictionary. Datasets, Options, and other types defined in this library implement subclass this ABC. This allows for polymorphic behavior when evaluating objects, and allows for third-party extensions to be created that can be used within the labrea framework.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

static unit(value: T) → Value[T]

Wrap a value in a Value object.

This method is used to wrap a value in a Value object. This is useful when you want to treat a value as an Evaluatable.

Parameters:: value (A) – The value to wrap.
Returns:: The wrapped value.
Return type:: Value[A]

static ensure(value: Evaluatable[T]) → Evaluatable[T]

static ensure(value: T) → Evaluatable[T]

Ensure that a value is an Evaluatable.

This method is used to ensure that a value is an Evaluatable. If the value is already an Evaluatable, it is returned as is. If the value is not an Evaluatable, it is wrapped in a Value object.

Parameters:: value (MaybeEvaluatable[A]) – The value to ensure is an Evaluatable.
Returns:: The value as an Evaluatable.
Return type:: Evaluatable[A]

apply(func: Evaluatable[Callable[[A], B]] | Callable[[A], B]) → Evaluatable[B]

Lazy application of a function to the result of evaluating the object.

This method is used to apply a function to the result of evaluating the object. The function is not evaluated until the object is evaluated. Equivalently can use the >> operator.

Parameters:: func (MaybeEvaluatable[Callable[[A], B]]) – The function to apply to the result of evaluating the object. Either a function of one argument, or an Evaluatable that evaluates to a function of one argument.
Returns:: A new evaluatable, that when evaluated, applies the function to the result of evaluating the source object.
Return type:: Evaluatable[B]

Example Usage

>>> from labrea import Option
>>>
>>> Option('A').apply(str.upper)({'A': 'foo'})
'FOO'
>>>
>>> (Option('B') >> str.upper)({'B': 'bar'})
'BAR'

bind(func: Callable[[A], Evaluatable[B]]) → Evaluatable[B]

Lazy bind a function to the result of evaluating the object.

This method is used to bind a function to the result of evaluating the object. The function should take the result of evaluating the object, and return an Evaluatable. This is useful for when you want to use the result of evaluating the object to determine the next step in the evaluation process.

Parameters:

func (Callable[[A], Evaluatable[B]]) – The function to bind to the result of evaluating the object.

Returns:

Evaluatable[B] – A new evaluatable, that when evaluated, binds the function to the result of evaluating the source object.
Example Usage
————-
>>> from labrea import Option
>>>
>>> x_if_a_neg_else_y = Option(‘A’).bind(lambda a (Option(‘X’) if a < 0 else Option(‘Y’)))
>>> x_if_a_neg_else_y({‘A’ (-1, ‘X’: ‘foo’, ‘Y’: ‘bar’}))
’foo’
>>> x_if_a_neg_else_y({‘A’ (1, ‘X’: ‘foo’, ‘Y’: ‘bar’}))
’bar’

property result: A

Dummy property used to appease type checkers.

When creating a dataset, type checkers will complain that the default values are not of the correct type. This property is used to tell the type checker that the default value is of the correct type.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.Value(value: A)

Bases: Evaluatable[A]

Simple wrapper for a plain value.

This class is used to wrap a value that is not an Evaluatable and make it an Evaluatable.

Parameters:: value (A) – The key to wrap.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.types.Apply(evaluatable: Evaluatable[A], func: Evaluatable[Callable[[A], B]])

Bases: Generic[A, B], Evaluatable[B]

A class representing the application of a function to the result of evaluating an object.

This class is used to apply a function to the result of evaluating an object. Apply objects are not usually created directly, instead the apply method is used on an Evaluatable object.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.types.Bind(evaluatable: Evaluatable[A], func: Callable[[A], Evaluatable[B]])

Bases: Generic[A, B], Evaluatable[B]

A class representing the binding of a function to the result of evaluating an object.

This class is used to bind a function to the result of evaluating an object. Bind objects are not usually created directly, instead the bind method is used on an Evaluatable object.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.types.Cacheable

Bases: ABC

Abstract base class for objects that can be cached.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

fingerprint(options: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | Sequence[str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]]) → bytes: Return a fingerprint, which is a unique identifier for a given evaluation.

class labrea.types.Validatable

Bases: ABC

Abstract base class for objects that can be validated against an options dictionary.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.types.Explainable

Bases: ABC

Abstract base class for objects that can explain themselves.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Options

Bases: Evaluatable[A]

A class representing a single user-provided option.

Options are the singular way to provide user-input to an Evaluatable object. They take a key, which is used to retrieve the value from the options dictionary. If the key does not exist, a default value can be provided.

Parameters:

key (str) – The key to retrieve from the options dictionary. This key can be nested using the standard {NESTED.KEY} syntax from confectioner.
default (MaybeEvaluatable[A]) – The default value to use if the key does not exist in the options dictionary. If the default value is an Evaluatable, it is evaluated using the options dictionary. If the default value is not an Evaluatable, it is returned as-is. If the default value is a string, it is treated as a Template and evaluated using the options dictionary.
default_factory (Callable[[], A]) – A factory function that returns the default value to use if the key does not exist in the options dictionary. This is an alternative to providing a default value directly.
doc (str) – The docstring for the option.
type (Type[A]) – The expected type of the option. Third-party packages can handle the labrea.type_validation.TypeValidationRequest to enforce types
domain (MaybeMissing[MaybeEvaluatable[_Domain]]) – A domain representing valid values for the option. The domain can be a predicate function, a container of valid values, or an Evaluatable that returns a predicate function or container of valid values (e.g. a pipeline step).

Example Usage

>>> from labrea import Option
>>> o = Option('A.X', default='foo')
>>> o()
'foo'
>>> o({'A': {'X': 'bar'}})
'bar'

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

static namespace(__namespace: Type) → Namespace

static namespace(__namespace: str) → Callable[[Type], Namespace]

Create an option namespace from a class definition

This allows all of the options for a module to be grouped together. Namespaces can contain Option objects or other namespaces.

When developing a package with Labrea, it is recommended to create an option namespace for with the same name as the package to avoid conflicts with other packages.

To create options with docstrings, see Option.auto().

Example Usage

>>> from labrea import Option
>>> @Option.namespace
... class MY_PACKAGE:
...     class MODULE_1:    # Implicit sub-namespace
...         A: str  # Equivalent to Option('MY_PACKAGE.MODULE_1.A')
...     @Option.namespace("MODULE-2")
...     class MODULE_2:    # Explicit sub-namespace with custom name
...         A = 10  # Equivalent to Option('MY_PACKAGE.MODULE_1.A', 10)
...
>>> MY_PACKAGE.MODULE_2.A()
10
>>> MY_PACKAGE.MODULE_2.A({'MY_PACKAGE': {'MODULE-2': {'A': 20}}})
20
>>> print(MY_PACKAGE.__doc__)
Namespace MY_PACKAGE:
    Namespace MY_PACKAGE.MODULE_1:
        Option MY_PACKAGE.MODULE_1.A
    Namespace MY_PACKAGE.MODULE-2:
        Option MY_PACKAGE.MODULE-2.A (default 10)

static auto(default: ~labrea._missing.Missing | ~labrea.types.Evaluatable[~labrea.option.A] | ~labrea.option.A = MISSING, doc: str = '', type: ~typing.Type[~labrea.option.A] = typing.Any, domain: ~labrea._missing.Missing | ~labrea.types.Evaluatable[~typing.Container[~labrea.option.A] | ~typing.Callable[[~labrea.option.A], bool]] | ~typing.Container[~labrea.option.A] | ~typing.Callable[[~labrea.option.A], bool] = MISSING) -> functools.partial(<class 'labrea.option.Option'>, type=+A)

Create an option in a namespace with an inferred key

Sometimes when creating a namespace, we want to add an option with a docstring or some additional transformations, but we want the key to be inferred from the namespace. This function creates an option with a default value and docstring, but automatically infers the key.

Example Usage

>>> from labrea import Option
>>> @Option.namespace
... class MY_PACKAGE:
...     A = Option.auto(default='foo', doc='An automatic option') >> str
...
>>> print(MY_PACKAGE.__doc__)
Namespace MY_PACKAGE:
    Option MY_PACKAGE.A (default 'foo'): An automatic option
>>> MY_PACKAGE.A()
'foo'
>>> MY_PACKAGE.A({'MY_PACKAGE': {'A': 100}})
'100'  # str transformation applied

Set the value of the option in the options dictionary.

Parameters:

options (Options) – The options dictionary to modify.
value (A) – The value to set the option to.

Returns:

The modified options dictionary

Return type:

Options

Example Usage

>>> from labrea import Option
>>> o = Option('A.Y')
>>> o.set({'A': {'X': 'foo'}}, 'bar')
{'A': {'X': 'foo', 'Y': 'bar'}}

Bases: Evaluatable[B]

A class that wraps an Evaluatable object and provides default options.

This class is used to provide default options to an Evaluatable object. The default options are mixed with the provided options. The default options can be forced to take precedence over the provided options (default) or the provided options can take precedence over the default options.

Parameters:

evaluatable (Evaluatable[B]) – The evaluatable object to wrap.
options (Options) – The default options to provide to the evaluatable object.
force (bool, optional) – If True, the default options take precedence over the provided options. If False, the provided options take precedence over the default options. Default is True.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Wrap an Evaluatable object with default options.

This is a convenience function for creating a WithOptions object with force=False.

Parameters:

evaluatable (Evaluatable[B]) – The evaluatable object to wrap.
options (Options) – The default options to provide to the evaluatable object.

Returns:

The wrapped evaluatable object with default options.

Return type:

WithOptions[B]

labrea.AllOptions: A special value that evaluates to the entire options dictionary.

Datasets

Bases: Evaluatable[A]

A class representing a dataset.

Datasets are the building blocks of Labrea programs. They represent the result of performing some calculation over a set of inputs. Datasets can be composed together to create complex call graphs that represent the flow of data through a program. Datasets are parameterized using Options, which allow for user input.

Datasets are not created directly; instead, they are created using the dataset decorator. This decorator takes a function that defines the dataset and returns a new Dataset object. When a dataset is evaluated, the default values to each argument are evaluated using the provided options. The dataset is then evaluated using the evaluated arguments.

Example Usage

>>> @dataset
... def a_squared(a: float = Option('A')) -> float:
...     return a ** 2
>>>
>>> @dataset
... def b_squared(b: float = Option('B')) -> float:
...     return b ** 2
>>>
>>> @dataset
... def hypotenuse(a2: float = a_squared, b2: float = b_squared) -> float:
...     return (a2 + b2) ** 0.5
>>>
>>> hypotenuse({'A': 3, 'B': 4})
5.0

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

overload(alias: Hashable | List[Hashable]) → Callable[[Callable[[...], A]], Dataset[A]]

Overloads the dataset with a new implementation. Used as a decorator.

Overloading a dataset allows you to provide a new implementation for the dataset. Overloads are selected based on the dispatch value provided to the dataset. If no dispatch value is provided, the default implementation is used. Datasets can also be abstract, in which no default implementation is provided, and an error is raised if no overload can be found.

Datasets must be created with a dispath argument in order to use overloads. If you are trying to overload a dataset that you do not own (i.e. from a third-party library), you can use my_dataset.set_dispatch(dispatch) to set the dispatch manually. The dispatch can be any Evaluatable object that returns a hashable value, or a string representing the Option key to use.

Parameters:: alias (Union[Hashable, List[Hashable]]) – The name or names to use for the overload.
Returns:: The Evaluatable object representing the new implementation.
Return type:: FunctionApplication[P, A]

Example Usage

>>> @dataset(dispatch='INPUT.SOURCE'))
... def input_data(path: str = Option('INPUT.PATH')) -> list[str]:
...     with open(path, 'r') as f:
...         return f.read()
>>>
>>> @input_data.overload('MOCK')
... def mock_input_data() -> list[str]:
...     return ['Mock', 'Data']
>>>
>>> input_data({'INPUT': {'PATH': 'data.txt'}})
['Data', 'From', 'File']
>>> input_data({'INPUT': {'SOURCE': 'MOCK'}})
['Mock', 'Data']

register(key: Hashable, value: Evaluatable[A]) → None

Registers a new overload for the dataset.

Registers a new overload for the dataset. Can be used if you want an overload to be a pre-existing Evaluatable object, like an Option.

Parameters:

key (Hashable) – The name to use for the overload.
value (Evaluatable[A]) – The Evaluatable object representing the new implementation.

Example Usage

>>> @dataset(dispatch='INPUT.SOURCE'))
... def input_data(path: str = Option('INPUT.PATH')) -> list[str]:
...     with open(path, 'r') as f:
...         return f.read()
>>>
>>> input_data.register('MOCK', Value(['Mock', 'Data']))

set_dispatch(dispatch: Evaluatable[Hashable]) → None

Sets the dispatch value for the dataset.

This is a stateful operation that changes the dispatch value for the dataset. This is intended for use with third-party datasets that do not have a dispatch value set, but you want to add an overload to.

Parameters:: dispatch (Evaluatable[Hashable]) – The dispatch value to use for the dataset.

set_cache(cache: Cache[A] | Callable[[...], Cache[A]]) → None

Sets the cache for the dataset.

Sets the cache for the dataset. The cache can be any object that implements the Cache interface, or a callable that returns a Cache object. This is a stateful operation that changes the cache for the dataset. This is intended for use with third-party datasets that do not have a cache set, but you want to add a cache to.

Parameters:: cache (Union[Cache[A], Callable[..., Cache[A]]]) – The cache to use for the dataset.

add_effects(*effects: Effect[A] | Evaluatable[Callable[[A], None]] | Callable[[A], None]) → None

Adds effects to the dataset.

Adds effects to the dataset. Effects are applied to the dataset when it is evaluated. This is a stateful operation that changes the effects for the dataset. This is intended for use with third-party datasets that you would like to add effects to (i.e. logging).

Parameters:: effects (Union[Effect[A], Callback[A]]) – The effects to add to the dataset.

add_effect(*effects: Effect[A] | Evaluatable[Callable[[A], None]] | Callable[[A], None]) → None

Adds effects to the dataset.

Adds effects to the dataset. Effects are applied to the dataset when it is evaluated. This is a stateful operation that changes the effects for the dataset. This is intended for use with third-party datasets that you would like to add effects to (i.e. logging).

Parameters:: effects (Union[Effect[A], Callback[A]]) – The effects to add to the dataset.

disable_effects() → None

Disables effects for the dataset.

Disables effects for the dataset. Effects are not executed when the dataset is evaluated. This is a stateful operation that changes the effects for the dataset. This is intended for use with third-party datasets that you would like to disable effects for.

enable_effects() → None

Enables effects for the dataset.

Enables effects for the dataset. Effects are executed when the dataset is evaluated. This is a stateful operation that changes the effects for the dataset. This is intended for use with third-party datasets that you would like to enable effects for.

Returns a new dataset with the provided options pre-set.

The provided options are pre-set, and cannot be overridden by the user. This can simplify the creation of multiple datasets that have the same inputs and calculations.

Parameters:: options (Options) – The options to pre-set for the dataset.
Returns:: A new dataset with the provided options pre-set.
Return type:: Dataset[A]

Example Usage

>>> @dataset
... def a_to_power(a: float = Option('A'), power: float = Option('POWER')) -> float:
...     return a ** power
>>>
>>> a_squared = a_to_power.with_options({'POWER': 2})

Returns a new dataset with the provided options as default.

The provided options are set as the default options for the dataset. These options can be overridden by the user. This can simplify the creation of multiple datasets that have the same inputs and calculations.

Parameters:: options (Options) – The options to set as the default for the dataset.
Returns:: A new dataset with the provided options as default.
Return type:: Dataset[A]

property default: Missing | Evaluatable[A]: The default implementation of the dataset.

property is_abstract: bool: Whether the dataset is abstract (i.e. has no default implementation).

labrea.dataset(definition: Callable[[...], A] | None = None, /, *, effects: List[Effect[A] | Evaluatable[Callable[[A], None]] | Callable[[A], None]] | None = None, cache: Cache[A] | Callable[[...], Cache[A]] | None = None, dispatch: Evaluatable[Hashable] | str | None = None, defaults: Dict[str, Any] | None = None, abstract: bool | None = None, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None, default_options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None, callback: Evaluatable[Callable[[A], A]] | Callable[[A], A] | None = None) → DatasetFactory[A] | Dataset[A]

labrea.abstractdataset(definition: Callable[[...], A] | None = None, /, *, effects: List[Effect[A] | Evaluatable[Callable[[A], None]] | Callable[[A], None]] | None = None, cache: Cache[A] | Callable[[...], Cache[A]] | None = None, dispatch: Evaluatable[Hashable] | str | None = None, defaults: Dict[str, Any] | None = None, abstract: bool | None = None, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None, default_options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None, callback: Evaluatable[Callable[[A], A]] | Callable[[A], A] | None = None) → DatasetFactory[A] | Dataset[A]

Conditionals

class labrea.Switch(dispatch: str | Evaluatable[Hashable], lookup: Mapping[Hashable, Evaluatable[V]], default: Missing | Evaluatable[V] | V = MISSING)

Bases: Evaluatable[V]

A class representing a switch statement.

This class takes a dispatch evaluatable and a mapping of keys to evaluatables. When evaluated, the dispatch is evaluated and the corresponding value is evaluated and returned. If the key is not found in the mapping, or the dispatch cannot be evaluated, a default value can be provided. If no default is provided, and the switch cannot choose a branch, an error is raised.

Aliases: labrea.switch(), labrea.Switch

Parameters:

dispatch (Union[str, Evaluatable[Hashable]]) – The dispatch evaluatable. This is used to determine which branch to take.
lookup (Mapping[Hashable, MaybeEvaluatable[V]]) – A mapping of keys to evaluatables. The key is determined by the dispatch. Values can also be plain values.
default (MaybeMissing[MaybeEvaluatable[V]], optional) – The default value to use if the key is not found in the mapping. Can be an evaluatable, or a plain value.

Example Usage

>>> from labrea import Option, switch
>>> s = switch(Option('A'), {True: Option('X'), False: Option('Y')}, Option('Z'))
>>> s({'A': True, 'X': 'Hello', 'Y': 'World', 'Z': 'Default'})
'Hello'
>>> s({'A': False, 'X': 'Hello', 'Y': 'World', 'Z': 'Default'})
'World'
>>> s({'A': None, 'X': 'Hello', 'Y': 'World', 'Z': 'Default'})
'Default'
>>> s({'X': 'Hello', 'Y': 'World', 'Z': 'Default'})
'Default'

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.conditional.CaseWhen(dispatch: Evaluatable[A], cases: Sequence[Tuple[Evaluatable[Callable[[A], bool]], Evaluatable[B]]], default: Missing | Evaluatable[B] = MISSING)

Bases: Generic[A, B], Evaluatable[B]

A class representing a case when statement.

This class takes a dispatch evaluatable and a sequence of cases. Each case is a condition and a result. When evaluated, the dispatch is evaluated and the corresponding case is matched. If no case is matched, a default value can be provided. If no default is provided, and the case cannot choose a branch, an error is raised.

This class is not usually instantiated directly. Instead, use the case function.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

when(condition: Evaluatable[Callable[[A], bool]] | Callable[[A], bool], result: Evaluatable[C]) → CaseWhen[A, B | C]

when(condition: Evaluatable[Callable[[A], bool]] | Callable[[A], bool], result: C) → CaseWhen[A, B | C]

Add a case to the case when statement. Returns a new instance.

Parameters:

condition (Union[str, Callable[[A], bool]]) – The condition to match. Can be a plain value or a callable.
result (MaybeEvaluatable[B]) – The result to return if the condition is matched.

otherwise(default: Evaluatable[C]) → CaseWhen[A, B | C]

otherwise(default: C) → CaseWhen[A, B | C]

Add a default value to the case when statement. Returns a new instance.

Parameters:: default (MaybeEvaluatable[B]) – The default value to return if no case is matched.

labrea.case(dispatch: Evaluatable[A]) → CaseWhen[A, Never]

labrea.case(dispatch: A) → CaseWhen[A, Never]

Create a case when statement.

This is the preferred method for creating a case when statement. It allows for chaining multiple cases and a default value.

Parameters:: dispatch (MaybeEvaluatable[A]) – The dispatch evaluatable. This is used to determine which branch to take.
Returns:: A new case when statement with no cases or default.
Return type:: CaseWhen[A, B]

Example Usage

>>> from labrea import Option, case
>>> c = case(
...     Option('A')
... ).when(
...     lambda a: a < 0,
...     Option('X')
... ).when(
...     lambda a: a > 0,
...     Option('Y')
... ).otherwise(Option('Z'))
>>> c({'A': -1, 'X': 'Negative', 'Y': 'Positive', 'Z': 'Zero'})
'Negative'
>>> c({'A': 1, 'X': 'Negative', 'Y': 'Positive', 'Z': 'Zero'})
'Positive'
>>> c({'A': 0, 'X': 'Negative', 'Y': 'Positive', 'Z': 'Zero'})
'Zero'

class labrea.Coalesce(_Coalesce__first: Evaluatable[A], *_Coalesce__rest: Evaluatable[A])

Bases: Evaluatable[A]

Return the first Evaluatable that can be evaluated

Takes 1 or more Evaluatables as arguments. Evaluates each Evaluatable in order until one can be evaluated. If none can be evaluated, raises an EvaluationError.

Aliases: labrea.coalesce(), labrea.Coalesce

Parameters:: *evaluatables (MaybeEvaluatable[A]) – The evaluatables to evaluate.

Example Usage

>>> from labrea import Coalesce, Option
>>> c = Coalesce(Option('A.X'), Option('B.Y'))
>>> c({'A': {'X': 1}, 'B': {'Y': 2}})
1
>>> c({'B': {'Y': 2}})
2

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Overloads

class labrea.Overloaded(dispatch: Evaluatable[Hashable], lookup: Dict[Hashable, Evaluatable[A]] | None, default: Missing | Evaluatable[A] | A = MISSING)

Bases: Evaluatable[A]

A class representing multiple implementations of an Evaluatable.

This class generalizes the idea of dataset overloads. It takes a dispatch, which is an evaluatable (like an Option) that returns a key, and a lookup dictionary, which maps keys to evaluatables. When the overloaded object is evaluated, it uses the dispatch to determine which evaluatable to use, and then evaluates that evaluatable. An optional default evaluatable can be provided, which is used if the dispatch key is not found in the lookup dictionary, or if the dispatch cannot be evaluated.

Parameters:

dispatch (Evaluatable[Hashable]) – The evaluatable that determines which implementation to use.
lookup (Dict[Hashable, Evaluatable[A]], optional) – A dictionary mapping keys to evaluatables.
default (MaybeMissing[MaybeEvaluatable[A]], optional) – The default evaluatable to use if the dispatch key is not found in the lookup dictionary. If this is not provided, an exception will be raised if the dispatch key is not found.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

register(key: Hashable, value: Evaluatable[A]) → None

Register a new implementation with the overloaded object.

Parameters:

key (Hashable) – The key for the implementation.
value (Evaluatable[A]) – The implementation to register.

property switch: Evaluatable[A]: Return the switch evaluatable that determines which implementation to use.

Interfaces

class labrea.interface.Interface(name, bases, namespace, dispatch: Evaluatable[Hashable])

Bases: type

Metaclass for interfaces.

Interfaces are classes that define a set of members that must be implemented by its implementations. The interface is defined by the members of the class, which are abstract datasets. The interface must have a dispatch, which is an evaluatable that returns a key that is used to determine which implementation to use.

Members of an interface are defined by the class attributes and annotations. Members can be defined implicitly using a type annotation, or explicitly using the @abstractdataset decorator. Members can also have a default implementation. Usually this is done by using the @dataset decorator, but any class attribute that is an Evaluatable can be used as a default implementation.

Normally, rather than using this metaclass directly, the @interface decorator should be used to create an interface.

implementation(alias: Hashable | List[Hashable]) → Callable[[T], T]

Decorator for implementing a single interface.

This is shorthand for using the @implements decorator with a single interface.

Parameters:: alias (Union[Hashable, List[Hashable]]) – The alias or aliases to register the implementation under.

class labrea.interface.Implementation(name, bases, namespace, interfaces: Tuple[Interface, ...], aliases: Tuple[Hashable, ...])

Bases: type

Metaclass for interface implementations.

Implementations are classes that implement one or more interfaces. Implementations are defined by the members of the class, which are datasets or other Evaluatables. When the implementation is created, the members are added as overloads to the corresponding members of the interfaces. If a member is not implemented, an error is raised.

Normally, rather than using this metaclass directly, the @implements decorator or the @MyInterface.implementation decorator should be used.

labrea.interface(dispatch: Evaluatable[Hashable] | str) → Callable[[T], T]

A decorator that creates an interface from a class.

An interface is a class that defines a set of members that must be implemented by its implementations. The interface is defined by the members of the class, which are abstract datasets. The interface must have a dispatch, which is an evaluatable that returns a key that is used to determine which implementation.

Downstream datasets can use interface members as if they were normal datasets. This allows developers to define a set of related datasets that must be implemented for a particular use case, enabling polymorphism and dependency injection.

Parameters:: dispatch (Evaluatable[Hashable] or str) – The evaluatable that determines which implementation to use. This can be an evaluatable (like an Option) or a string representing an Option.
Returns:: A decorator that creates an interface from a class.
Return type:: Callable[[T], T]

Example Usage

>>> from labrea import interface, abstractdataset, dataset, Option
>>> @interface("DISPATCH.KEY")
... class MyInterface:
...     a: float  # Implicit member
...
...     # Explicit member
...     @abstractdataset
...     def b() -> str:
...         pass
...
...     # Member with default implementation
...     @dataset
...     def c() -> str:
...         return 'C'
...
...     # Member with default implementation, not using @dataset
...     d: str = Option('D')
>>>
>>> @dataset
... def a_squared(a: float = MyInterface.a) -> float:
...     return a ** 2
>>>

labrea.implements(*interfaces: Interface, alias: Hashable | List[Hashable] | None = None) → Callable[[T], T]

Decorator used for implementing one or more interfaces at once.

Parameters:

*interfaces (Interface) – The interfaces to implement.
alias (Union[Hashable, List[Hashable], None]) – The alias or aliases to register the implementation under. Keyword-only argument; required.

Returns:

A decorator that creates an implementation from a class.

Return type:

Callable[[T], T]

Example Usage

>>> @implements(Interface1, Interface2, alias="ALIAS")
... class MyImplementation:
...     # Static value member
...     a: str = 'A'
...     # Non-dataset Evaluatable member
...     b: str = Option('B')
...     # Explicit dataset member
...     @dataset
...     def c() -> str:
...         return 'C'
...     # Implicit dataset member
...     def d() -> str:
...         return 'D'

Iterables & Collections

class labrea.Iter(*evaluatables: Evaluatable[A])

Bases: Evaluatable[Iterable[A]]

A class representing multiple evaluatables as an iterable.

This is used to lazily evaluate multiple evaluatables in a single operation. The result is an iterable of the evaluated values.

Parameters:: *evaluatables (MaybeEvaluatable[A]) – The evaluatables to evaluate.

Example Usage

>>> from labrea import Iter, Option
>>> i = Iter(Option('A.X'), Option('B.Y'))
>>> for value in i({'A': {'X': 1}, 'B': {'Y': 2}}):
...     print(value)
1
2

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

labrea.evaluatable_list(*evaluatables: Evaluatable[A]) → Evaluatable[List[A]]

Create an Evaluatable that evaluates to a list of values.

This function is a convenience wrapper around Iter.apply(list).

Aliases: labrea.DatasetList(), labrea.evaluatable_list()

Parameters:: *evaluatables (Evaluatable[A]) – The objects to evaluate.

labrea.evaluatable_tuple(__e1: Evaluatable[T1]) → Evaluatable[Tuple[T1]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2]) → Evaluatable[Tuple[T1, T2]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3]) → Evaluatable[Tuple[T1, T2, T3]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3], __e4: Evaluatable[T4]) → Evaluatable[Tuple[T1, T2, T3, T4]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3], __e4: Evaluatable[T4], __e5: Evaluatable[T5]) → Evaluatable[Tuple[T1, T2, T3, T4, T5]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3], __e4: Evaluatable[T4], __e5: Evaluatable[T5], __e6: Evaluatable[T6]) → Evaluatable[Tuple[T1, T2, T3, T4, T5, T6]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3], __e4: Evaluatable[T4], __e5: Evaluatable[T5], __e6: Evaluatable[T6], __e7: Evaluatable[T7]) → Evaluatable[Tuple[T1, T2, T3, T4, T5, T6, T7]]

labrea.evaluatable_tuple(__e1: Evaluatable[T1], __e2: Evaluatable[T2], __e3: Evaluatable[T3], __e4: Evaluatable[T4], __e5: Evaluatable[T5], __e6: Evaluatable[T6], __e7: Evaluatable[T7], __e8: Evaluatable[T8]) → Evaluatable[Tuple[T1, T2, T3, T4, T5, T6, T7, T8]]

Create an Evaluatable that evaluates to a tuple of values.

This function is a convenience wrapper around Iter.apply(tuple).

Aliases: labrea.DatasetTuple(), labrea.evaluatable_tuple()

Parameters:: *evaluatables (Evaluatable[A]) – The objects to evaluate.

labrea.evaluatable_set(*evaluatables: Evaluatable[K]) → Evaluatable[Set[K]]

Create an Evaluatable that evaluates to a set of values.

This function is a convenience wrapper around Iter.apply(set).

Aliases: labrea.DatasetSet(), labrea.evaluatable_set()

Parameters:: *evaluatables (Evaluatable[K]) – The objects to evaluate.

labrea.evaluatable_dict(contents: Dict[K, Evaluatable[V]]) → Evaluatable[Dict[K, V]]

Create an Evaluatable that evaluates to a dictionary of values.

This function is a convenience wrapper around Iter.apply(dict).

Aliases: labrea.DatasetDict(), labrea.evaluatable_dict()

Parameters:: contents (Dict[K, Evaluatable[V]]) – The objects to evaluate.

A class that represents the same evaluatable repeated over multiple options.

This is used to evaluate a single evaluatable multiple times with different options. The result is an iterable of tuples, where the first element is the options used to evaluate the evaluatable, and the second element is the evaluated value. This is the Labrea equivalent of a for loop.

Parameters:

evaluatable (Evaluatable[A]) – The evaluatable to evaluate.
iterables (Dict[str, MaybeEvaluatable[Iterable[JSON]]]) – A dictionary mapping option keys to iterables of options. The keys can be dotted to represent nested options.

Example Usage

>>> from labrea import Map, Option, dataset
>>> @dataset
... def hypotenuse(a: int = Option('A'), b: int = Option('B')) -> float:
...     return (a ** 2 + b ** 2) ** 0.5
>>>
>>> hypotenuses = Map(hypotenuse, {'A': Option('X'), 'B': Option('Y')})
>>> for options, value in hypotenuses({'X': [3, 4], 'Y': [4, 5]}):
...     print(options, value)
{'A': 3, 'B': 4} 5.0
{'A': 3, 'B': 5} 5.830951894845301
{'A': 4, 'B': 4} 5.656854249492381
{'A': 4, 'B': 5} 6.403124237432848

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

property values: Evaluatable[Iterable[A]]

Strip the options from the results.

This is a convenience method that returns an evaluatable that returns the second element of the tuples returned by the Map evaluatable.

Pipelines

class labrea.pipeline.PipelineStep(step: Evaluatable[Callable[[A], B]], _name: str | None = None)

Bases: Evaluatable[Callable[[A], B]], Transformation[A, B]

A class representing a single step in a pipeline.

A pipeline step is a single transformation that can be applied to a value. Steps can be composed into pipelines using the + operator.

This class should probably not be used directly. Instead, use the pipeline_step

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Transform a value using the pipeline step.

Parameters:

value (A) – The value to transform.
options (Optional[Options]) – The options to use when transforming the value.

Returns:

The transformed value.

Return type:

B

class labrea.pipeline.Pipeline

class labrea.pipeline.Pipeline(tail: PipelineStep[B, C])

class labrea.pipeline.Pipeline(tail: PipelineStep[B, C], rest: Pipeline[A, B])

Bases: Evaluatable[Callable[[A], B]], Iterable[Evaluatable[Callable[[Any], Any]]], Transformation[A, B]

A class representing a pipeline of transformations.

A pipeline is a sequence of transformations that can be applied to a value. Pipelines are implemented as a linked list of PipelineSteps, where each step is a single transformation that can be applied to a value. Pipelines, like PipelineSteps, can be composed using the + operator.

This class should probably not be used directly. Instead, use the pipeline_step and compose pipelines using the + operator.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

transform(value: A, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → B: Transform a value using the pipeline.

labrea.pipeline_step(func: Callable[[Concatenate[A, P]], B]) → PipelineStep[A, B]

Create a pipeline step from a function.

This is the primary way to create a pipeline step. The function should be used as a decorator on the function that will be used as the transformation. The function should take at least one argument. The first argument will be the value to transform; the remaining arguments will be the parameters for the transformation, and should have default values similar to a dataset definition.

Example Usage

>>> @pipeline_step
... def add(x: int, y: int = Option('AMOUNT', 1)) -> int:
...     return x + y
>>>
>>> @pipeline_step
... def multiply(x: int, y: int = Option('FACTOR', 2)) -> int:
...     return x * y
>>>
>>> pipeline = add + multiply
>>> pipeline.transform(1)
4
>>> pipeline.transform(1, {'AMOUNT': 2, 'FACTOR': 3})
9

Pipeline Helper Functions

Generic

labrea.functions.partial(__func: ~labrea.types.Evaluatable[~typing.Callable[[~P], ~labrea.functions.A]] | ~typing.Callable[[~P], ~labrea.functions.A], *args: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → Evaluatable[Callable[[...], A]]

Analog of functools.partial that works with Evaluatables.

Parameters:

func (MaybeEvaluatable[Callable[P, A]]) – The function to partially apply.
*args (MaybeEvaluatable[P.args]) – The positional arguments to partially apply to the function. These can be Evaluatables that return values, or constant values.
**kwargs (MaybeEvaluatable[P.kwargs]) – The keyword arguments to partially apply to the function. These can be Evaluatables that return values, or constant values.

Returns:

An Evaluatable that returns a partially applied function.

Return type:

Evaluatable[Callable[…, A]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.partial(lambda x, y: x + y, y=Option('B')))({'A': 1, 'B': 2})
3

labrea.functions.into(func: Evaluatable[Callable[[...], A]] | Callable[[...], A]) → PipelineStep[Iterable | Mapping[str, Any], A]

Convert a function that takes positional arguments into one that takes an iterable.

This can be useful if you have an evaluatable that returns an iterable of args or a mapping of kwargs and you want to unpack that into a function that takes positional or keyword arguments.

Parameters:: func (MaybeEvaluatable[Callable[..., A]]) – The function to apply to the arguments. Can be an evaluatable that returns a function or a constant function.
Returns:: A pipeline step that applies the function to the arguments after unpacking.
Return type:: PipelineStep[Union[Iterable, Mapping[str, Any], A]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.into(lambda x, y: x + y))({'A': [1, 2]})
3

labrea.functions.ensure(__predicate: Evaluatable[Callable[[A], bool]] | Callable[[A], bool], __msg: Missing | Evaluatable[str] | str = MISSING) → PipelineStep[A, A]

Create a pipeline step that checks if the input satisfies a predicate.

Parameters:

__predicate (MaybeEvaluatable[Callable[[A], bool]]) – The predicate to check if the input satisfies. This can be an Evaluatable that returns a predicate, or a constant predicate.
__msg (MaybeMissing[MaybeEvaluatable[str]], optional) – The message to use if the predicate fails. This can be an Evaluatable that returns a message, or a constant message. If not provided, a default message will be used.

Returns:

A pipeline step that checks if the input satisfies the predicate.

Return type:

PipelineStep[A, A]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.ensure(lambda x: x > 0))({'A': 1})
1

labrea.functions.get_attribute(__name: Evaluatable[str] | str) → PipelineStep[Any, Any]

Create a pipeline step that gets an attribute from the input.

Parameters:: __name (MaybeEvaluatable[str]) – The name of the attribute to get. This can be an Evaluatable that returns a string, or a constant string.
Returns:: A pipeline step that gets the attribute from the input.
Return type:: PipelineStep[Any, Any]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.get_attribute('upper'))({'A': 'hello'})
'HELLO'

labrea.functions.call_method(__name: Evaluatable[str] | str, *args: Any, **kwargs: Any) → PipelineStep[Any, Any]

Create a pipeline step that calls a method on the input.

Parameters:

__name (MaybeEvaluatable[str]) – The name of the method to call. This can be an Evaluatable that returns a string, or a constant string.
*args (Any) – The positional arguments to pass to the method.
**kwargs (Any) – The keyword arguments to pass to the method.

Returns:

A pipeline step that calls the method on the input.

Return type:

PipelineStep[Any, Any]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.call_method('upper'))({'A': 'hello'})
'HELLO'

Collections

labrea.functions.map(func: Evaluatable[Callable[[A], B]] | Callable[[A], B]) → PipelineStep[Iterable[A], Iterable[B]]

Create a pipeline step that maps a function over an iterable.

This allows for calling map in a functional style over an iterable using the .apply() method or the >> operator on an Evaluatable object that returns an iterable.

Parameters:: func (MaybeEvaluatable[Callable[[A], B]]) – The function to apply to each element of the iterable. This can be an Evaluatable that returns a function, or a constant function.

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.map(lambda x: x + 1) >> list)({'A': [1, 2, 3, 4]})
[2, 3, 4, 5]

labrea.functions.filter(func: Evaluatable[Callable[[A], bool]] | Callable[[A], bool]) → PipelineStep[Iterable[A], Iterable[A]]

Create a pipeline step that filters an iterable using a predicate function.

This allows for calling filter in a functional style over an iterable using the .apply() method or the >> operator on an Evaluatable object that returns an iterable.

Parameters:: func (MaybeEvaluatable[Callable[[A], bool]]) – The predicate function to apply to each element of the iterable. This can be an Evaluatable that returns a function, or a constant function.

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.filter(lambda x: x % 2 == 0) >> list)({'A': [1, 2, 3, 4]})
[2, 4]

labrea.functions.reduce(func: Evaluatable[Callable[[A, B], A]] | Callable[[A, B], A], initial: Missing | Evaluatable[A] | A = MISSING) → PipelineStep[Iterable[B], A]

Create a pipeline step that reduces an iterable using a binary function.

This allows for calling reduce in a functional style over an iterable using the .apply() method or the >> operator on an Evaluatable object that returns an iterable.

Parameters:

func (MaybeEvaluatable[Callable[[A, A], A]]) – The binary function to apply to each element of the iterable. This can be an Evaluatable that returns a function, or a constant function.
initial (MaybeMissing[MaybeEvaluatable[A]], optional) – The initial value to use for the reduction. If not provided, the first element of the iterable is used as the initial value. This can be an Evaluatable that returns a value, or a constant value.

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.reduce(lambda x, y: x + y))({'A': [1, 2, 3, 4]})
10

labrea.functions.flatten(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: Create a pipeline step that flattens an iterable of iterables.

labrea.functions.flatmap(func: Evaluatable[Callable[[A], Iterable[B]]] | Callable[[A], Iterable[B]]) → PipelineStep[Iterable[A], Iterable[B]]

Create a pipeline step that maps a function over an iterable and flattens the result.

This allows for calling flatmap in a functional style over an iterable using the .apply() method or the >> operator on an Evaluatable object that returns an iterable.

Parameters:: func (MaybeEvaluatable[Callable[[A], Iterable[B]]]) – The function to apply to each element of the iterable. This can be an Evaluatable that returns a function, or a constant function.

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.flatmap(lambda x: [x, x + 1]) >> list)({'A': [1, 2, 3, 4]})
[1, 2, 2, 3, 3, 4, 4, 5]

labrea.functions.map_items(func: Evaluatable[Callable[[K1, V1], Tuple[K2, V2]]] | Callable[[K1, V1], Tuple[K2, V2]]) → PipelineStep[Mapping[K1, V1], Mapping[K2, V2]]

Create a pipeline step that maps a function over the items of a mapping.

Parameters:: func (MaybeEvaluatable[Callable[[K1, V1], Tuple[K2, V2]]]) – The function to apply to each item of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that applies the function to the items of the mapping.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K2, V2]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.map_items(lambda k, v: (k + 1, v + 1)))({'A': {1: 2, 3: 4}})
mappingproxy({2: 3, 4: 5})

labrea.functions.map_keys(func: Evaluatable[Callable[[K1], K2]] | Callable[[K1], K2]) → PipelineStep[Mapping[K1, V1], Mapping[K2, V1]]

Create a pipeline step that maps a function over the keys of a mapping.

Parameters:: func (MaybeEvaluatable[Callable[[K1], K2]]) – The function to apply to each key of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that applies the function to the keys of the mapping.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K2, V1]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.map_keys(lambda k: k + 1))({'A': {1: 2, 3: 4}})
mappingproxy({2: 2, 4: 4})

labrea.functions.map_values(func: Evaluatable[Callable[[V1], V2]] | Callable[[V1], V2]) → PipelineStep[Mapping[K1, V1], Mapping[K1, V2]]

Create a pipeline step that maps a function over the values of a mapping.

Parameters:: func (MaybeEvaluatable[Callable[[V1], V2]]) – The function to apply to each value of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that applies the function to the values of the mapping.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K1, V2]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.map_values(lambda v: v + 1))({'A': {1: 2, 3: 4}})
mappingproxy({1: 3, 3: 5})

labrea.functions.filter_items(func: Evaluatable[Callable[[K1, V1], bool]] | Callable[[K1, V1], bool]) → PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Create a pipeline step that filters items from a mapping using a predicate function.

Parameters:: func (MaybeEvaluatable[Callable[[K1, V1], bool]]) – The predicate function to apply to each item of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that filters the items of the mapping using the predicate function.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.filter_items(lambda k, v: k % 2 == 0))({'A': {1: 2, 3: 4}})
mappingproxy({3: 4})

labrea.functions.filter_keys(func: Evaluatable[Callable[[K1], bool]] | Callable[[K1], bool]) → PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Create a pipeline step that filters keys from a mapping using a predicate function.

Parameters:: func (MaybeEvaluatable[Callable[[K1], bool]]) – The predicate function to apply to each key of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that filters the keys of the mapping using the predicate function.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.filter_keys(lambda k: k % 2 == 0))({'A': {1: 2, 3: 4}})
mappingproxy({1: 2})

labrea.functions.filter_values(func: Evaluatable[Callable[[V1], bool]] | Callable[[V1], bool]) → PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Create a pipeline step that filters values from a mapping using a predicate function.

Parameters:: func (MaybeEvaluatable[Callable[[V1], bool]]) – The predicate function to apply to each value of the mapping. This can be an Evaluatable that returns a function, or a constant function.
Returns:: A pipeline step that filters the values of the mapping using the predicate function.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K1, V1]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.filter_values(lambda v: v % 2 == 0))({'A': {1: 2, 3: 4}})
mappingproxy({1: 2})

labrea.functions.concat(iterable: Evaluatable[Iterable[A]] | Iterable[A]) → PipelineStep[Iterable[B], Iterable[A | B]]

Create a pipeline step that concatenates an iterable to another iterable.

Parameters:: iterable (MaybeEvaluatable[Iterable[A]]) – The iterable to concatenate to the input. This can be an Evaluatable that returns an iterable, or a constant iterable.
Returns:: A pipeline step that concatenates the input with the iterable.
Return type:: PipelineStep[Iterable[A], Iterable[A]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.append([4, 5, 6]))({'A': [1, 2, 3]})
[1, 2, 3, 4, 5, 6]

labrea.functions.append(item: Evaluatable[A] | A) → PipelineStep[Iterable[B], Iterable[A | B]]

Create a pipeline step that appends an item to an iterable.

Parameters:: item (MaybeEvaluatable[A]) – The item to append to the input. This can be an Evaluatable that returns an item, or a constant item.
Returns:: A pipeline step that appends the item to the input.
Return type:: PipelineStep[Iterable[A], Iterable[A]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.append(4))({'A': [1, 2, 3]})
[1, 2, 3, 4]

labrea.functions.intersect(collection: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], Set[H]]

Create a pipeline step that intersects an iterable with another collection.

Parameters:: collection (MaybeEvaluatable[Iterable[H]]) – The collection to intersect with the input. This can be an Evaluatable that returns a collection, or a constant collection.
Returns:: A pipeline step that intersects the input with the collection.
Return type:: PipelineStep[Iterable[H], Set[H]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.intersect([1, 2, 3]))({'A': [2, 3, 4]})
{2, 3}

labrea.functions.union(collection: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], Set[H]]

Create a pipeline step that unions an iterable with another collection.

Parameters:: collection (MaybeEvaluatable[Iterable[H]]) – The collection to union with the input. This can be an Evaluatable that returns a collection, or a constant collection.
Returns:: A pipeline step that unions the input with the collection.
Return type:: PipelineStep[Iterable[H], Set[H]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.union([1, 2, 3]))({'A': [2, 3, 4]})
{1, 2, 3, 4}

labrea.functions.difference(collection: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], Set[H]]

Create a pipeline step that computes the difference between sets.

Parameters:: collection (MaybeEvaluatable[Iterable[H]]) – The collection to compute the difference with the input. This can be an Evaluatable that returns a collection, or a constant collection.
Returns:: A pipeline step that computes the difference between the input and the collection.
Return type:: PipelineStep[Iterable[H], Set[H]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.difference([1, 2, 3]))({'A': [2, 3, 4]})
{4}

labrea.functions.symmetric_difference(collection: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], Set[H]]

Create a pipeline step that computes the symmetric difference between sets.

Parameters:: collection (MaybeEvaluatable[Iterable[H]]) – The collection to compute the symmetric difference with the input. This can be an Evaluatable that returns a collection, or a constant collection.
Returns:: A pipeline step that computes the symmetric difference between the input and the collection.
Return type:: PipelineStep[Iterable[H], Set[H]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.symmetric_difference([1, 2, 3]))({'A': [2, 3, 4]})
{1, 4}

labrea.functions.merge(mapping: Evaluatable[Mapping[K1, V1]] | Mapping[K1, V1]) → PipelineStep[Mapping[K2, V2], Mapping[K1 | K2, V1 | V2]]

Create a pipeline step that merges a mapping with another mapping.

Parameters:: mapping (MaybeEvaluatable[Mapping[K1, K2]]) – The mapping to merge with the input. This can be an Evaluatable that returns a mapping, or a constant mapping.
Returns:: A pipeline step that merges the input with the mapping.
Return type:: PipelineStep[Mapping[K1, V1], Mapping[K2, V1]]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.merge({'A': 'B'}))({'A': {'A': 1, 'B': 2}})
{'B': 1}

labrea.functions.get(__x: Evaluatable[X] | X, default: Missing = MISSING) → PipelineStep[_Indexable[X, Y], Y]

labrea.functions.get(__x: Evaluatable[X] | X, default: Z) → PipelineStep[_Indexable[X, Y], Y | Z]

Create a pipeline step that gets a value from the input at the given key/index

Parameters:

__x (X) – The key/index to get the value from the input. This can be an Evaluatable that returns a value, or a constant value.
default (MaybeMissing[Y], optional) – The default value to return if the key/index is not found in the input. This can be an Evaluatablethat returns a value, or a constant value. If not provided, an exception will be raised if the key/index is not found.

Returns:

A pipeline step that gets the value from the input at the given key/index.

Return type:

PipelineStep[_Indexable[X, Y], Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.get(1))({'A': ['a', 'b', 'c'])]
'b'

labrea.functions.get_from(__x: Evaluatable[_Indexable[X, Y]] | _Indexable[X, Y], default: Missing) → PipelineStep[X, Y]

labrea.functions.get_from(__x: Evaluatable[_Indexable[X, Y]] | _Indexable[X, Y], default: Z) → PipelineStep[X, Y | Z]

Create a pipeline step that gets a value from the input at the given key/index from the left.

This will reverse the operand order compared to get.

Parameters:

__x (_Indexable[X, Y]) – The key/index to get the value from the input. This can be an Evaluatable that returns a value, or a constant value.
default (MaybeMissing[Z], optional) – The default value to return if the key/index is not found in the input. This can be an Evaluatable

Returns:

A pipeline step that gets the value from the input at the given key/index.

Return type:

PipelineStep[X, Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.get_from(['a', 'b', 'c']))({'A': 1})
'b'

labrea.functions.length(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: Create a pipeline step that computes the length of a sequence.

Math

labrea.functions.add(__x: Evaluatable[_Addable[X, Y]] | _Addable[X, Y]) → PipelineStep[X, Y]

Create a pipeline step that adds a value to the input.

Parameters:: __x (_Addable[X, Y]) – The value to add to the input. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that adds the value to the input.
Return type:: PipelineStep[X, Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.add(1))({'A': 2})
3

labrea.functions.subtract(__x: Evaluatable[_Subtractable[X, Y]] | _Subtractable[X, Y]) → PipelineStep[X, Y]

Create a pipeline step that subtracts a value from the input.

Parameters:: __x (_Subtractable[X, Y]) – The value to subtract from the input. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that subtracts the value from the input.
Return type:: PipelineStep[X, Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.subtract(1))({'A': 2})
1

labrea.functions.multiply(__x: Evaluatable[_Multiplicable[X, Y]] | _Multiplicable[X, Y]) → PipelineStep[X, Y]

Create a pipeline step that multiplies the input by a value.

Parameters:: __x (_Multiplicable[X, Y]) – The value to multiply the input by. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that multiplies the input by the value.
Return type:: PipelineStep[X, Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.multiply(2))({'A': 3})
6

labrea.functions.left_multiply(__x: Evaluatable[X] | X) → PipelineStep[_Multiplicable[X, Y], Y]

Create a pipeline step that multiplies the input by a value from the left.

This will reverse the operand order compared to multiply, which is useful when multiplacation is not commutative.

Parameters:: __x (X) – The value to multiply the input by. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that multiplies the input by the value.
Return type:: PipelineStep[_Multiplicable[X, Y], Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.multiply_by(2))({'A': 3})
6

labrea.functions.divide_by(__x: Evaluatable[_Divisible[X, Y]] | _Divisible[X, Y]) → PipelineStep[X, Y]

Create a pipeline step that divides the input by a value.

Parameters:: __x (_Divisible[X, Y]) – The value to divide the input by. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that divides the input by the value.
Return type:: PipelineStep[X, Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.divide(2))({'A': 6})
3

labrea.functions.divide_into(__x: Evaluatable[X] | X) → PipelineStep[_Divisible[X, Y], Y]

Create a pipeline step that divides the input into a value from the left.

This will reverse the operand order compared to divide.

Parameters:: __x (X) – The value to divide the input into. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that divides the input by the value.
Return type:: PipelineStep[_Divisible[X, Y], Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.divide_into(2))({'A': 6})
3

labrea.functions.negate(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: Pipeline step that negates the input.

labrea.functions.modulo(__x: Evaluatable[X] | X) → PipelineStep[_Modable[X, Y], Y]

Create a pipeline step that computes the modulo of the input with a value.

Parameters:: __x (_Modable[X, Y]) – The value to compute the modulo with. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that computes the modulo of the input with the value.
Return type:: PipelineStep[_Modable[X, Y], Y]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.modulo(2))({'A': 5})
1

Predicates

labrea.functions.eq(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is equal to a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is equal to the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.eq(1))({'A': 1})
True

labrea.functions.ne(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is not equal to a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is not equal to the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.ne(1))({'A': 1})
False

labrea.functions.lt(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is less than a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is less than the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.lt(1))({'A': 0})
True

labrea.functions.le(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is less than or equal to a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is less than or equal to the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.le(1))({'A': 1})
True

labrea.functions.gt(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is greater than a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is greater than the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.gt(1))({'A': 2})
True

labrea.functions.ge(value: Any) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is greater than or equal to a value.

Parameters:: value (Any) – The value to compare the input to.
Returns:: A pipeline step that checks if the input is greater than or equal to the value.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.ge(1))({'A': 1})
True

labrea.functions.positive(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is positive.

labrea.functions.negative(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is negative.

labrea.functions.non_positive(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is non-positive.

labrea.functions.non_negative(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is non-negative.

labrea.functions.is_none(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is None.

labrea.functions.is_not_none(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is not None.

labrea.functions.has_remainder(divisor: Evaluatable[X] | X, reminder: Evaluatable[Y] | Y) → PipelineStep[_Modable[X, Y], bool]

Create a pipeline step that checks if the input has a remainder when divided by a divisor.

Parameters:

divisor (MaybeEvaluatable[int]) – The divisor to divide the input by. This can be an Evaluatable that returns a divisor, or a constant divisor.
reminder (MaybeEvaluatable[int]) – The reminder to check if the input has when divided by the divisor. This can be an Evaluatable that returns a reminder, or a constant reminder.

Returns:

A pipeline step that checks if the input has a reminder when divided by the divisor.

Return type:

PipelineStep[int, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.has_remainder(2, 1))({'A': 3})
True

labrea.functions.even(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is even.

labrea.functions.odd(options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → A: A pipeline step that checks if the input is odd.

labrea.functions.any(*funcs: Evaluatable[Callable[[Any], bool]] | Callable[[Any], bool]) → PipelineStep[Any, bool]

Create a pipeline step that checks if any functions return True for the input.

Parameters:: *funcs (MaybeEvaluatable[Callable[[Any], bool]]) – The functions to apply to the input. These can be Evaluatables that return functions, or constant functions.
Returns:: A pipeline step that checks if any functions return True for the input.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.any(lambda x: x < 0, lambda x: x > 10))({'A': 5})
False

labrea.functions.all(*funcs: Evaluatable[Callable[[Any], bool]] | Callable[[Any], bool]) → PipelineStep[Any, bool]

Create a pipeline step that checks if all functions return True for the input.

Parameters:: *funcs (MaybeEvaluatable[Callable[[Any], bool]]) – The functions to apply to the input. These can be Evaluatables that return functions, or constant functions.
Returns:: A pipeline step that checks if all functions return True for the input.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.all(lambda x: x > 0, lambda x: x < 10))({'A': 5})
True

labrea.functions.invert(func: ~labrea.types.Evaluatable[~typing.Callable[[~typing.Any], bool]] | ~typing.Callable[[~typing.Any], bool] = <function <lambda>>) → PipelineStep[Any, bool]

Create a pipeline step that negates the result of a function.

Parameters:: func (MaybeEvaluatable[Callable[[Any], bool]], optional) – The function to negate the result of. This can be an Evaluatable that returns a function, or a constant function. If omitted, the identity function is used
Returns:: A pipeline step that negates the result of the function.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.invert(lambda x: x < 0))({'A': 5})
True
>>> (Option('A') >> F.invert())({'A': True})
False

labrea.functions.instance_of(*types: Evaluatable[type] | type) → PipelineStep[Any, bool]

Create a pipeline step that checks if the input is an instance of a type.

Parameters:: *types (MaybeEvaluatable[type]) – The types to check if the input is an instance of. These can be Evaluatables that return types, or constant types.
Returns:: A pipeline step that checks if the input is an instance of the type.
Return type:: PipelineStep[Any, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.instance_of(int))({'A': 1})
True

labrea.functions.is_in(container: Evaluatable[Container[A]] | Container[A]) → PipelineStep[A, bool]

Create a pipeline step that checks if the input is in a container.

Parameters:: container (MaybeEvaluatable[Container[A]]) – The container to check if the input is in. This can be an Evaluatable that returns a container, or a constant container.
Returns:: A pipeline step that checks if the input is in the container.
Return type:: PipelineStep[A, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.is_in([1, 2, 3]))({'A': 2})
True

labrea.functions.is_not_in(container: Evaluatable[Container[A]] | Container[A]) → PipelineStep[A, bool]

Create a pipeline step that checks if the input is not in a container.

Parameters:: container (MaybeEvaluatable[Container[A]]) – The container to check if the input is in. This can be an Evaluatable that returns a container, or a constant container.
Returns:: A pipeline step that checks if the input is not in the container.
Return type:: PipelineStep[A, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.is_in([1, 2, 3]))({'A': 4})
True

labrea.functions.one_of(*items: Evaluatable[A] | A) → PipelineStep[A, bool]

Create a pipeline step that checks if the input is one of the items.

Parameters:: *items (MaybeEvaluatable[A]) – The items to check if the input is one of. These can be Evaluatables that return items, or constant items.
Returns:: A pipeline step that checks if the input is one of the items.
Return type:: PipelineStep[A, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.one_of(1, 2, 3))({'A': 2})
True

labrea.functions.none_of(*items: Evaluatable[A] | A) → PipelineStep[A, bool]

Create a pipeline step that checks if the input is none of the items.

Parameters:: *items (MaybeEvaluatable[A]) – The items to check if the input is one of. These can be Evaluatables that return items, or constant items.
Returns:: A pipeline step that checks if the input is none of the items.
Return type:: PipelineStep[A, bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.none_of(1, 2, 3))({'A': 4})
True

labrea.functions.contains(value: Evaluatable[A] | A) → PipelineStep[Container[A], bool]

Create a pipeline step that checks if the container contains a value.

Parameters:: value (MaybeEvaluatable[A]) – The value to check if the container contains. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that checks if the container contains the value.
Return type:: PipelineStep[Container[A], bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.contains(2))({'A': [1, 2, 3]})
True

labrea.functions.does_not_contain(value: Evaluatable[A] | A) → PipelineStep[Container[A], bool]

Create a pipeline step that checks if the container does not contain a value.

Parameters:: value (MaybeEvaluatable[A]) – The value to check if the container contains. This can be an Evaluatable that returns a value, or a constant value.
Returns:: A pipeline step that checks if the container does not contain the value.
Return type:: PipelineStep[Container[A], bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.does_not_contain(4))({'A': [1, 2, 3]})
True

labrea.functions.intersects(iterable: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], bool]

Create a pipeline step that checks if the iterable intersects with another iterable.

Parameters:: container (MaybeEvaluatable[Iterable[H]]) – The iterable to check if the input intersects with. This can be an Evaluatable that returns a iterable, or a constant iterable.
Returns:: A pipeline step that checks if the input intersects with the iterable.
Return type:: PipelineStep[Iterable[H], bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.intersects([1, 2, 3]))({'A': [2, 3, 4]})
True

labrea.functions.disjoint_from(iterable: Evaluatable[Iterable[H]] | Iterable[H]) → PipelineStep[Iterable[H], bool]

Create a pipeline step that checks if the iterable is disjoint from another iterable.

Parameters:: container (MaybeEvaluatable[Iterable[A]]) – The iterable to check if the input is disjoint from. This can be an Evaluatable that returns a iterable, or a constant iterable.
Returns:: A pipeline step that checks if the input is disjoint from the iterable.
Return type:: PipelineStep[Iterable[A], bool]

Example Usage

>>> from labrea import Option
>>> import labrea.functions as F
>>>
>>> (Option('A') >> F.disjoint_from([1, 2, 3]))({'A': [4, 5, 6]})
True

Templates

class labrea.Template(template: str, **kwargs)

Bases: Evaluatable[str]

A template string that can be evaluated using other Evaluatables.

Template entries use the format {:key:} where key is the name of a parameter (keyword argument). The parameter can be any Evaluatable. The template is evaluated using the options dictionary and the result is returned as a string. Normal options can also be used in the template using the standard {NESTED.KEY} syntax from confectioner.

Parameters:

template (str) – The template string to evaluate
kwargs (Any) – The parameters to use when evaluating the template. Each parameter can be any Evaluatable. If a parameter is not an Evaluatable, it is used as a constant key.

Raises:

ValueError – If the template requires parameters that are not provided

Example Usage

>>> from labrea import Template, dataset, Option
>>> @dataset
... def b_dataset(b: str = Option('B')) -> str:
...     return b
>>>
>>> t = Template('{A.X} {:b:}', b=b_dataset)
>>> t({'A': {'X': 'Hello'}, 'B': 'World!'})
'Hello World!'

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

DatasetClasses

labrea.datasetclass(c: Type[A]) → _DatasetClassMeta[A]

Create a new DatasetClass from a class definition.

DatasetClasses are classes that when instatiated, evaluate their members using the options dict provided to the constructor. This allows for the definition of complex data structures that can be evaluated at runtime.

Any members of the class that are not Evaluatables are wrapped in labrea.Value instances.

Example Usage

>>> from labrea import dataset, datasetclass, Option
>>> @dataset
... def my_dataset(a: str = Option('A')) -> str:
...     return a
>>>
>>> @datasetclass
... class MyDatasetClass:
...     a: str = my_dataset
...     b: int = Option('B')
...     c: bool = True
...
>>> inst = MyDatasetClass({'A': 'Hello World!', 'B': 1})
>>> print(inst.a, inst.b, inst.c)
Hello World! 1 True

Function Application

class labrea.arguments.Arguments(*args: ~typing.~P, **kwargs: ~typing.~P)

Bases: Generic[P]

A class representing the arguments of a function or method.

Parameters:

*args (P.args) – The positional arguments.
**kwargs (P.kwargs) – The keyword arguments.

class labrea.arguments.EvaluatableArgs(*args: ~labrea.types.Evaluatable[~typing.~P])

Bases: Generic[P], Evaluatable[P.args]

A class representing a set of arguments that can be evaluated.

Parameters:: *args (Evaluatable[P.args]) – The arguments to evaluate.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.arguments.EvaluatableKwargs(**kwargs: ~labrea.types.Evaluatable[~typing.~P])

Bases: Generic[P], Evaluatable[P.kwargs]

A class representing a set of keyword arguments that can be evaluated.

Parameters:: **kwargs (Evaluatable[P.kwargs]) – The keyword arguments to evaluate.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.arguments.EvaluatableArguments(*args: ~labrea.types.Evaluatable[~typing.~P], **kwargs: ~labrea.types.Evaluatable[~typing.~P])

Bases: Evaluatable[Arguments[P]]

A class representing a set of arguments that can be evaluated.

Parameters:

*args (MaybeEvaluatable[P.args]) – The positional arguments to evaluate.
**kwargs (MaybeEvaluatable[P.kwargs]) – The keyword arguments to evaluate

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

labrea.arguments.arguments(*args: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → Evaluatable[Arguments[P]]

Create an EvaluatableArguments object from the given arguments.

Parameters:

*args (MaybeEvaluatable[P.args]) – The positional arguments to evaluate.
**kwargs (MaybeEvaluatable[P.kwargs]) – The keyword arguments to evaluate

class labrea.application.FunctionApplication(_FunctionApplication__func: ~labrea.types.Evaluatable[~typing.Callable[[~P], ~labrea.application.A]] | ~typing.Callable[[~P], ~labrea.application.A], /, *args: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P)

Bases: Generic[P, A], Evaluatable[A]

A class representing the application of a function to a set of arguments.

This class is what is used by the labrea.dataset.Dataset under the hood.

Parameters:

__func (Callable[P, A]) – The function to apply.
*args (MaybeEvaluatable[P.args]) – The positional arguments to evaluate.
**kwargs (MaybeEvaluatable[P.kwargs]) – The keyword arguments to evaluate.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

classmethod lift(__func: ~typing.Callable[[~P], ~labrea.application.A], /, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → FunctionApplication[P, A]

classmethod lift(__func: None = None, /, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → Callable[[Callable[[P], A]], FunctionApplication[P, A]]

Lift a function definition with Evaluatable default arguments to a FunctionApplication.

This is used by the labrea.dataset() decorator under the hood. Can be used as a decorator with or without keyword arguments for the defaults.

Parameters:

__func (Callable[P, A]) – The function to apply.
**kwargs (MaybeEvaluatable[P.kwargs]) – Default values for the keyword arguments of the function.

Returns:

The lifted function application.

Return type:

FunctionApplication[P, A]

Example Usage

>>> @FunctionApplication.lift
... def a_squared(a: int = Option('A')) -> int:
...     return a ** 2
>>>
>>> @FunctionApplication(a=Option('A'))
... def a_squared(a: int) -> int:
...     return a ** 2
>>>
>>> a_squared = FunctionApplication.lift(lambda a: a**2, a=Option('A'))

class labrea.application.PartialApplication(_PartialApplication__func: ~labrea.types.Evaluatable[~typing.Callable[[~P], ~labrea.application.A]] | ~typing.Callable[[~P], ~labrea.application.A], /, *args: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P)

Bases: Generic[P, A], Evaluatable[Callable[[…], A]]

A class representing the partial application of a function to a set of arguments.

This class is used by the labrea.pipeline_step() decorator under the hood.

Parameters:

__func (Callable[P, A]) – The function to apply.
*args (MaybeEvaluatable[P.args]) – The positional arguments to evaluate.
**kwargs (MaybeEvaluatable[P.kwargs]) – The keyword arguments to evaluate.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

classmethod lift(__func: ~typing.Callable[[~P], ~labrea.application.A], /, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → PartialApplication[P, A]

classmethod lift(__func: None = None, /, **kwargs: ~labrea.types.Evaluatable[~typing.~P] | ~typing.~P) → Callable[[Callable[[P], A]], PartialApplication[P, A]]

Lift a function definition with Evaluatable default arguments to a PartialApplication.

This is used by the labrea.pipeline_step() decorator under the hood. Can be used as a decorator with or without keyword arguments for the defaults.

Parameters:

__func (Callable[P, A]) – The function to apply.
**kwargs (MaybeEvaluatable[P.kwargs]) – Default values for the keyword arguments of the function.

Returns:

The lifted partial application.

Return type:

PartialApplication[P, A]

Example Usage

>>> @PartialApplication.lift
... def a_plus_b(a: int, b: int = Option('B')) -> int:
...     return a + b
>>>
>>> @PartialApplication(b=Option('B'))
... def a_plus_b(a: int, b: int) -> int:
...     return a + b
>>>
>>> a_plus_b = PartialApplication.lift(lambda a, b: a + b, b=Option('B'))

Runtime

class labrea.runtime.Request

Bases: Generic[A]

Base class for all requests.

Requests are objects that are passed to the runtime to be handled by a handler. The handler is responsible for processing the request and returning a result. Requests are used to represent side effects using data, and allow the runtime to execute the side effects in a controlled manner.

run() → A

Runs the request using the current runtime, returning the result.

This method should not be overridden.

classmethod handle(handler: Callable[[Self], A]) → Callable[[Self], A]

Decorator to register a default handler for a request type.

This method should not be overridden.

Parameters:: handler (Handler[Self, A]) – The handler to register for the request type.
Returns:: The handler that was registered.
Return type:: Handler[Self, A]

Example Usage

>>> class PrintRequest(Request[None]):
...     msg: str
...     def __init__(self, msg: str, options: Options) -> None:
...         self.msg = msg
...         self.options = options
>>>
>>> @MyRequest.handle
... def print_handler(request: MyRequest) -> str:
...     print(request.msg)
>>>
>>> PrintRequest("Hello, world!", {}).run()
Hello, world!

class labrea.runtime.Runtime(handlers: Mapping[Type[Request], Callable[[R], A]] | None = None)

Bases: object

A class that manages the execution of requests via handlers.

Runtimes are used to execute side effects in a controlled manner. They maintain a mapping of request types to handlers, and execute the appropriate handler for each request.

Can be used as a context manager, which sets the current runtime for the duration of the context block for the current thread.

Parameters:: handlers (Optional[Mapping[Type[Request], Handler]]) – A mapping of request types to handlers. If provided, these handlers will be used in place of the default handlers.

handle(request: Type[R] | Mapping[Type[Request], Callable[[R], A]], handler: Callable[[R], A] | None = None) → Runtime

Creates a new runtime with the provided request type and handler.

This method returns a new runtime with the provided request type and handler registered. The new runtime will inherit the handlers from the current runtime, and will override any handlers for the provided request type.

Parameters:

request (Union[Type[R], Mapping[Type[Request], Handler]]) – The request type to handle, or a mapping of request types to handlers.
handler (Optional[Handler[R, A]]) – The handler to register for the request type.

Returns:

A new runtime with the provided request type and handler registered.

Return type:

Runtime

Raises:

TypeError – If the request is not a request type or a mapping of request types to handlers.

run(request: Request[A]) → A

Runs the provided request using the registered handler.

This is not meant to be called directly. Instead, use the run() method on the request object.

Parameters:: request (Request[A]) – The request to run.
Returns:: The result of running the request with the correct handler.
Return type:: A

labrea.runtime.current_runtime() → Runtime: Returns the current runtime for the current thread.

labrea.runtime.handle(request: Type[R] | Mapping[Type[Request], Callable[[R], A]], handler: Callable[[R], A] | None = None) → Runtime

Alias for current_runtime().handle(request, handler).

This is useful when using the runtime as a context manager.

Parameters:

request (Union[Type[R], Mapping[Type[Request], Handler]]) – The request type to handle, or a mapping of request types to handlers.
handler (Optional[Handler[R, A]]) – The handler to register for the request type.

Returns:

A new runtime with the provided request type and handler registered

Return type:

Runtime

Example Usage

>>> with handle(MyRequest, my_handler):
...     MyRequest("Hello, world!", {}).run()

labrea.runtime.handle_by_default(request: Type[R], handler: Callable[[R], A]) → None

Registers a default handler for a request type.

Parameters:

request (Type[R]) – The request type to handle.
handler (Handler[R, A]) – The handler to register for the request type.

labrea.runtime.inherit(parent: Thread) → None

Inherit the runtime from a parent thread.

Parameters:: parent (threading.Thread) – The parent thread to inherit the runtime from.

Computation

class labrea.computation.Effect(*args, **kwargs)

Bases: Transformation[A, None], Validatable, Explainable, ABC

Abstract base class for effects.

Effects are transformations that are applied to values, but do not return a new value. They are intended to encapsulate side effects, such as logging, data validation, or other operations that do not affect the value being returned by an Evaluatable.

Apply the effect to a value.

Parameters:

value (A) – The value to apply the effect to.
options (Options) – The options dictionary to evaluate against.

Returns:

The value after the effect has been applied.

Return type:

A

Raises:

EvaluationError – If the effect cannot be applied to the value.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.computation.ChainedEffect(*effects: Effect[A])

Bases: Effect[A]

An effect that chains multiple effects together.

This effect applies a sequence of effects to a value in order.

Parameters:: *effects (Effect) – The effects to chain together.

transform(value: A, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → None: Perform each effect in sequence.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.computation.CallbackEffect(callback: Evaluatable[Callable[[A], None]] | Callable[[A], None])

Bases: Effect[A]

An effect that applies a callback to a value.

This allows any function of one value to be used as an effect. The callback can also be an Evaluatable that returns a function, which allows Pipeline objects to be used as effects.

Parameters:: callback (MaybeEvaluatable[Callable[[A], None]]) – The callback to apply to the value. Either a function of one value that returns None, or an Evaluatable that returns such a function.

transform(value: A, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → None: Apply the callback to the value.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.computation.Computation(evaluatable: Evaluatable[A], effect: Effect[A])

Bases: Evaluatable[A]

A computation that applies an effect to a value.

This class combines an Evaluatable with an Effect to create a new Evaluatable that applies the effect to the value returned by the original Evaluatable. This allows side effects to be applied to values without modifying the original Evaluatable.

If the LABREA.EFFECTS.DISABLED option is set to True, the effect will not be applied to the value. This allows the effect to be disabled in certain contexts, such as testing or debugging.

Parameters:

evaluatable (Evaluatable[A]) – The Evaluatable to apply the effect to.
effect (Effect[A]) – The Effect to apply to the value returned by the Evaluatable.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

Caching

class labrea.cache.Cached(evaluatable: Evaluatable[A], cache: Cache[A])

Bases: Evaluatable[A]

A class representing an Evaluatable that may be cached.

When evaluated, this class will check if the value exists in the cache before evaluating the underlying Evaluatable. If the value is not in the cache, the value will be evaluated and stored in the cache before being returned.

Parameters:

evaluatable (Evaluatable) – The evaluatable to cache.
cache (Cache) – The cache to store the value in.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

labrea.cached(__x: Evaluatable[A], cache: Cache[A] | None = None) → Evaluatable[A]

labrea.cached(__x: Cache[A]) → Callable[[Evaluatable[A]], Evaluatable[A]]

Return an Evaluatable that caches the result of evaluating the input Evaluatable.

Can be used as a decorator or a function. If used as a decorator, can be used with or without providing a cache object. If no cache object is provided, a MemoryCache will be used.

Parameters:

__x (Union[Cache, Evaluatable]) – The cache object or evaluatable to cache.
cache (Optional[Cache], optional) – The cache object to use, by default None, in which case a MemoryCache is used.

Returns:

The cached evaluatable, or a function that takes an evaluatable and returns a cached evaluatable.

Return type:

Union[Callable[[Evaluatable[A]], Evaluatable[A]], Evaluatable[A]]

Example Usage

>>> from labrea import cached, Option
>>> from labrea.cache import MemoryCache
>>> from labrea.application import FunctionApplication
>>> x_cached = cached(Option('X'))
>>> x_cached_with_cache = cached(Option('X'), MemoryCache())
>>>
>>> @cached
... @FunctionApplication.lift
... def y(x: int = Option('X')) -> int:
...     return x
>>>
>>> @cached(MemoryCache())
... @FunctionApplication.lift
... def z(x: int = Option('X')) -> int:
...     return x

class labrea.cache.Cache

Bases: Generic[A], ABC

A class representing a cache of values that can be evaluated.

Get a key from the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to get from the cache.
options (Options) – The options dictionary to evaluate against.

Returns:

The value from the cache.

Return type:

A

Raises:

CacheGetFailure – If the key cannot be retrieved from the cache.

Set a key in the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to set in the cache.
options (Options) – The options dictionary to evaluate against.
value (A) – The value to set in the cache.

Raises:

CacheSetFailure – If the key cannot be set in the cache.

Check if a key exists in the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to check in the cache.
options (Options) – The options dictionary to evaluate against.

Returns:

Whether the value exists in the cache.

Return type:

bool

Raises:

CacheExistsFailure – If the existence of the key cannot be checked in the cache.

class labrea.cache.MemoryCache

Bases: Cache[A]

A class representing a cache that stores values in memory.

Get a key from the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to get from the cache.
options (Options) – The options dictionary to evaluate against.

Returns:

The value from the cache.

Return type:

A

Raises:

CacheGetFailure – If the key cannot be retrieved from the cache.

Set a key in the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to set in the cache.
options (Options) – The options dictionary to evaluate against.
value (A) – The value to set in the cache.

Raises:

CacheSetFailure – If the key cannot be set in the cache.

Check if a key exists in the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to check in the cache.
options (Options) – The options dictionary to evaluate against.

Returns:

Whether the value exists in the cache.

Return type:

bool

Raises:

CacheExistsFailure – If the existence of the key cannot be checked in the cache.

class labrea.cache.NoCache

Bases: Cache[Any]

A class representing a cache that does not store any values.

Get a key from the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to get from the cache.
options (Options) – The options dictionary to evaluate against.

Returns:

The value from the cache.

Return type:

A

Raises:

CacheGetFailure – If the key cannot be retrieved from the cache.

Set a key in the cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to set in the cache.
options (Options) – The options dictionary to evaluate against.
value (A) – The value to set in the cache.

Raises:

CacheSetFailure – If the key cannot be set in the cache.

Bases: Request[A]

A request to set a value in a cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable that produced the value.
options (Options) – The options dictionary that was used to evaluate the value.
value (A) – The value to set in the cache.
cache (Cache) – The cache to set the value in.

Bases: Request[A]

A request to get a value from a cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to get from the cache.
options (Options) – The options dictionary to evaluate against.
cache (Cache) – The cache to get the value from.

Bases: Request[bool]

A request to check if a value exists in a cache.

Parameters:

evaluatable (Evaluatable) – The evaluatable to check in the cache.
options (Options) – The options dictionary to evaluate against.
cache (Cache) – The cache to check the value in.

Logging

Bases: Request[None]

A request to log a message.

Parameters:

level (int) – The logging level (see the logging module).
name (str) – The name of the logger; usually the module name.
msg (str) – The message to log.
options (Options) – The options used during evaluation

class labrea.logging.Logged(evaluatable: Evaluatable[A], level: int, name: str, msg: str, log_first: bool = True)

Bases: Evaluatable[A]

An Evaluatable that logs a message before (or after) evaluating another Evaluatable.

Parameters:

evaluatable (Evaluatable[A]) – The Evaluatable to evaluate.
level (int) – The logging level (see the logging module).
name (str) – The name of the logger; usually the module name.
msg (str) – The message to log.
log_first (bool, optional) – Whether to log the message before or after evaluating the Evaluatable. Default is True.

Evaluate the object.

Parameters:: options (Options) – The options dictionary to evaluate against.
Returns:: The evaluated value.
Return type:: A
Raises:: EvaluationError – If the object cannot be evaluated.

Notes

This method should be called recursively on all child objects where applicable.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return the keys from the Options that this object depends on.

This should only return the keys that are present in the options dictionary. If a key is not present but is required, a KeyNotFoundError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

class labrea.logging.LogEffect(level: int, name: str, msg: str)

Bases: Effect

An effect that logs a message.

Parameters:

level (int) – The logging level (see the logging module).
name (str) – The name of the logger; usually the module name.
msg (str) – The message to log.

transform(value: None, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]] | None = None) → None: Log the message.

Validate the object.

Parameters:: options (Options) – The options dictionary to validate against.
Raises:: KeyNotFoundError – If a required key is not found in the options dictionary.

Notes

This method should be called recursively on all child objects where applicable.

Return all keys that this object depends on.

This should return all keys that this object depends on, including those that are not present in the options dictionary. If the keys required cannot be determined, an InsufficientInformationError should be raised.

Parameters:: options (Options) – The options dictionary to validate against.
Returns:: The keys that this object depends on.
Return type:: Set[str]
Raises:: InsufficientInformationError – If the keys required cannot be determined.

Notes

This method should be called recursively on all child objects where applicable.

labrea.logging.disabled() → Runtime

Return a runtime that disables logging.

Can be used as a context manager to disable logging for the duration of the context block.

Example Usage

>>> import labrea.logging
>>> with labrea.logging.disabled():
...     pass

labrea.logging.DEBUG(name: str, msg: str, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]) → None: Log a message at the DEBUG level.

labrea.logging.INFO(name: str, msg: str, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]) → None: Log a message at the INFO level.

labrea.logging.WARNING(name: str, msg: str, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]) → None: Log a message at the WARNING level.

labrea.logging.ERROR(name: str, msg: str, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]) → None: Log a message at the ERROR level.

labrea.logging.CRITICAL(name: str, msg: str, options: Mapping[str, str | int | float | bool | None | Mapping[str, JSON] | Sequence[JSON]]) → None: Log a message at the CRITICAL level.

Exceptions

class labrea.exceptions.EvaluationError(msg: str, source: Evaluatable)

Bases: Exception

Error raised when an object cannot be evaluated.

class labrea.exceptions.KeyNotFoundError(key: str, source: Evaluatable)

Bases: EvaluationError

Error raised when a key is not found in the options dictionary.

class labrea.exceptions.InsufficientInformationError(reason: str, source: Evaluatable)

Bases: EvaluationError

Error raised when not enough information is provided to explain an object.