Data processors

zenpyre.data_processors

Contain data processors.

zenpyre.data_processors.BaseProcessor

Bases: ABC, Generic[U, T]

Abstract base class for data processors.

A processor takes an input of type U and returns an output of type T. Unlike :class:~zenpyre.ingestors.base.BaseIngestor, which fetches data from a configured source, a processor always receives its input explicitly via :meth:process. This makes processors composable — the output of one can be passed directly as the input of another.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
U		The input type accepted by :meth:process.	required
T		The output type returned by :meth:process.	required

Subclasses must implement :meth:process.

Example

>>> from zenpyre.data_processors import BaseProcessor
>>> class UpperCaseProcessor(BaseProcessor[str, str]):
...     def process(self, data: str) -> str:
...         return data.upper()
...
>>> UpperCaseProcessor().process("hello")
'HELLO'

zenpyre.data_processors.BaseProcessor.process abstractmethod

process(data: U) -> T

Process the input data and return the result.

Parameters:

Name	Type	Description	Default
data	U	The input data to process.	required

Returns:

Type	Description
T	The processed output. Its type and structure are defined
T	by the concrete implementation.

zenpyre.data_processors.FilterDocumentsByMetadataProcessor

Bases: BaseProcessor[list[Document], list[Document]], InlineDisplayMixin

Processor that filters a list of LangChain documents by the value of a metadata key.

Wraps :func:~zenpyre.documents.filter_by_metadata as a :class:~zenpyre.data_processors.base.BaseProcessor so it can be composed in a :class:~zenpyre.data_processors.SequentialProcessor pipeline.

Parameters:

Name	Type	Description	Default
metadata_key	str	The metadata key to filter by.	required
value	Any	The value to match against. Documents whose metadata_key equals this value are kept.	required

Example

>>> from langchain_core.documents import Document
>>> from zenpyre.data_processors import FilterDocumentsByMetadataProcessor
>>> processor = FilterDocumentsByMetadataProcessor(metadata_key="category", value="Science")
>>> docs = [
...     Document(page_content="A", metadata={"category": "Science"}),
...     Document(page_content="B", metadata={"category": "Cooking"}),
...     Document(page_content="C", metadata={"category": "Science"}),
... ]
>>> result = processor.process(docs)
>>> [doc.page_content for doc in result]
['A', 'C']

zenpyre.data_processors.FilterDocumentsByMetadataProcessor.process

process(data: list[Document]) -> list[Document]

Filter data by the configured metadata key and value.

Parameters:

Name	Type	Description	Default
data	list[Document]	The list of :class:~langchain_core.documents.Document instances to filter.	required

Returns:

Type	Description
list[Document]	A new list containing only the
list[Document]	class:~langchain_core.documents.Document instances whose
list[Document]	metadata_key equals value. The original list is
list[Document]	not modified.

zenpyre.data_processors.FilterDocumentsByMetadataRangeProcessor

Bases: BaseProcessor[list[Document], list[Document]], InlineDisplayMixin

Processor that filters a list of LangChain documents by a range of values for a metadata key.

Wraps :func:~zenpyre.documents.filter_by_metadata_range as a :class:~zenpyre.data_processors.base.BaseProcessor so it can be composed in a :class:~zenpyre.data_processors.SequentialProcessor pipeline.

Parameters:

Name	Type	Description	Default
metadata_key	str	The metadata key to filter by.	required
lower	Any	The inclusive lower bound. Pass None (the default) for no lower bound.	None
upper	Any	The inclusive upper bound. Pass None (the default) for no upper bound.	None

Example

>>> from langchain_core.documents import Document
>>> from zenpyre.data_processors import FilterDocumentsByMetadataRangeProcessor
>>> processor = FilterDocumentsByMetadataRangeProcessor(
...     metadata_key="page", lower=2, upper=4
... )
>>> docs = [
...     Document(page_content="A", metadata={"page": 1}),
...     Document(page_content="B", metadata={"page": 3}),
...     Document(page_content="C", metadata={"page": 5}),
... ]
>>> result = processor.process(docs)
>>> [doc.page_content for doc in result]
['B']

zenpyre.data_processors.FilterDocumentsByMetadataRangeProcessor.process

process(data: list[Document]) -> list[Document]

Filter data by the configured metadata key and range.

Parameters:

Name	Type	Description	Default
data	list[Document]	The list of :class:~langchain_core.documents.Document instances to filter.	required

Returns:

Type	Description
list[Document]	A new list containing only the
list[Document]	class:~langchain_core.documents.Document instances whose
list[Document]	metadata_key value falls within [lower, upper].
list[Document]	The original list is not modified.

zenpyre.data_processors.FilterDocumentsByMetadataValuesProcessor

Bases: BaseProcessor[list[Document], list[Document]], InlineDisplayMixin

Processor that filters a list of LangChain documents by checking if a metadata value is in a set of accepted values.

Wraps :func:~zenpyre.documents.filter_by_metadata_values as a :class:~zenpyre.data_processors.base.BaseProcessor so it can be composed in a :class:~zenpyre.data_processors.SequentialProcessor pipeline.

Parameters:

Name	Type	Description	Default
metadata_key	str	The metadata key to filter by.	required
values	set[Any]	The set of accepted values. Documents whose metadata_key is in this set are kept.	required

Example

>>> from langchain_core.documents import Document
>>> from zenpyre.data_processors import FilterDocumentsByMetadataValuesProcessor
>>> processor = FilterDocumentsByMetadataValuesProcessor(
...     metadata_key="category", values={"Science", "Technology"}
... )
>>> docs = [
...     Document(page_content="A", metadata={"category": "Science"}),
...     Document(page_content="B", metadata={"category": "Cooking"}),
...     Document(page_content="C", metadata={"category": "Technology"}),
... ]
>>> result = processor.process(docs)
>>> sorted(doc.page_content for doc in result)
['A', 'C']

zenpyre.data_processors.FilterDocumentsByMetadataValuesProcessor.process

process(data: list[Document]) -> list[Document]

Filter data by the configured metadata key and set of values.

Parameters:

Name	Type	Description	Default
data	list[Document]	The list of :class:~langchain_core.documents.Document instances to filter.	required

Returns:

Type	Description
list[Document]	A new list containing only the
list[Document]	class:~langchain_core.documents.Document instances whose
list[Document]	metadata_key value is in values. The original list
list[Document]	is not modified.

zenpyre.data_processors.FirstNProcessor

Bases: BaseProcessor[Sequence[T], list[T]], InlineDisplayMixin

Processor that returns the first n items from a sequence.

Parameters:

Name	Type	Description	Default
n	int	The maximum number of items to return. Must be a positive integer.	required

Raises:

Type	Description
ValueError	If n is not a positive integer.

Example

>>> from zenpyre.data_processors import FirstNProcessor
>>> processor = FirstNProcessor(n=2)
>>> processor.process([1, 2, 3, 4, 5])
[1, 2]
>>> processor.process([1])
[1]

zenpyre.data_processors.FirstNProcessor.process

process(data: Sequence[T]) -> list[T]

Return the first n items from data.

Parameters:

Name	Type	Description	Default
data	Sequence[T]	The input sequence to slice.	required

Returns:

Type	Description
list[T]	A list containing the first n elements of data, or
list[T]	all elements if len(data) < n.

zenpyre.data_processors.LambdaProcessor

Bases: BaseProcessor[U, T], InlineDisplayMixin

Processor that applies a callable to its input and returns the result.

Unlike :class:~zenpyre.data_processors.LambdaSequenceProcessor, which applies fn to each item of a sequence independently, this class calls fn once on the whole input. Useful for wrapping any single-value transformation in a pipeline without writing a dedicated processor class.

Parameters:

Name	Type	Description	Default
fn	Callable[[U], T]	A callable that takes the input of type U and returns a value of type T.	required

Example

>>> from zenpyre.data_processors import LambdaProcessor
>>> processor = LambdaProcessor(fn=len)
>>> processor.process(["a", "b", "c"])
3
>>> processor = LambdaProcessor(fn=sorted)
>>> processor.process([3, 1, 2])
[1, 2, 3]

zenpyre.data_processors.LambdaProcessor.process

process(data: U) -> T

Apply fn to data and return the result.

Parameters:

Name	Type	Description	Default
data	U	The input to process.	required

Returns:

Type	Description
T	The output of fn applied to data.

zenpyre.data_processors.LastNProcessor

Bases: BaseProcessor[Sequence[T], list[T]], InlineDisplayMixin

Processor that returns the last n items from a sequence.

Parameters:

Name	Type	Description	Default
n	int	The maximum number of items to return. Must be a positive integer.	required

Raises:

Type	Description
ValueError	If n is not a positive integer.

Example

>>> from zenpyre.data_processors import LastNProcessor
>>> processor = LastNProcessor(n=2)
>>> processor.process([1, 2, 3, 4, 5])
[4, 5]
>>> processor.process([1])
[1]

zenpyre.data_processors.LastNProcessor.process

process(data: Sequence[T]) -> list[T]

Return the last n items from data.

Parameters:

Name	Type	Description	Default
data	Sequence[T]	The input sequence to slice.	required

Returns:

Type	Description
list[T]	A list containing the last n elements of data, or
list[T]	all elements if len(data) < n.

zenpyre.data_processors.SequenceProcessor

Bases: BaseProcessor[Sequence[U], list[T]], MultilineDisplayMixin

Processor that applies another processor to each item in a sequence and returns the list of results.

Useful for applying any :class:~zenpyre.data_processors.base.BaseProcessor to every element of a sequence without writing a dedicated class. Unlike :class:~zenpyre.data_processors.LambdaProcessor, which applies processor to the whole input at once, this class maps processor over each item individually.

Parameters:

Name	Type	Description	Default
processor	BaseProcessor[U, T]	A :class:~zenpyre.data_processors.base.BaseProcessor that accepts a single item of type U and returns a value of type T.	required
progress_description	str	The description shown on the progress bar while processing. Defaults to "Processing items...".	'Processing items...'
raise_on_error	bool	If True (default), an exception raised while processing an item is propagated and processing stops immediately. If False, items that fail are logged and skipped, and processing continues with the remaining items.	True
max_workers	int	The number of worker threads to use to process items concurrently. Defaults to 0, which processes items sequentially in the calling thread (no thread pool is created in that case). Any value >= 1 processes items concurrently using a pool of max_workers threads.	0

Example

>>> from zenpyre.data_processors import LambdaProcessor, SequenceProcessor
>>> processor = SequenceProcessor(processor=LambdaProcessor(fn=str.upper))
>>> processor.process(["hello", "world"])
['HELLO', 'WORLD']
>>> processor = SequenceProcessor(processor=LambdaProcessor(fn=len))
>>> processor.process(["a", "bb", "ccc"])
[1, 2, 3]

zenpyre.data_processors.SequenceProcessor.process

process(data: Sequence[U]) -> list[T]

Apply processor to each item in data and return the results.

Parameters:

Name	Type	Description	Default
data	Sequence[U]	The sequence of items to process.	required

Returns:

Type	Description
list[T]	A list of results in the same order as the successfully
list[T]	processed items in data. If raise_on_error is
list[T]	False, items that raise an exception are skipped and
list[T]	omitted from the returned list.

zenpyre.data_processors.SequentialProcessor

Bases: BaseProcessor[Any, Any]

Processor that applies a sequence of processors one after another, passing the output of each as the input to the next.

Mirrors the design of :class:torch.nn.Sequential: processors are composed in the order they are provided, and the final output is the result of the last processor. If no processors are provided, :meth:process returns the input unchanged.

Parameters:

Name	Type	Description	Default
*processors	BaseProcessor[Any, Any]	Zero or more :class:~zenpyre.data_processors.base.BaseProcessor instances to apply in order.	()

Example

>>> from zenpyre.data_processors import LambdaProcessor, SequentialProcessor
>>> p = SequentialProcessor(
...     LambdaProcessor(fn=lambda x: x * 2),
...     LambdaProcessor(fn=str),
... )
>>> p.process(21)
'42'
>>> SequentialProcessor().process(42)
42

zenpyre.data_processors.SequentialProcessor.process

process(data: Any) -> Any

Apply each processor in sequence and return the final result.

Passes data to the first processor, then feeds each processor's output as the input of the next.

Parameters:

Name	Type	Description	Default
data	Any	The initial input data passed to the first processor.	required

Returns:

Type	Description
Any	The output of the last processor in the sequence.

zenpyre.data_processors.ShuffleProcessor

Bases: BaseProcessor[Sequence[T], list[T]], MultilineDisplayMixin

Processor that shuffles the items of a sequence.

Takes an iterable (e.g. ``list``, ``tuple``) as input and returns
a new ``list`` with the same elements in random order. The result
is always materialized as a new ``list``, so the original sequence
type of the input is not preserved.

The shuffling uses its own :class:`random.Random` instance, so it
does not affect and is not affected by the global ``random``
module state. Note that the internal random generator is created
once, in ``__init__``, and its state advances with each call to
:meth:`process`. This means calling ``process`` multiple times on
the same instance — even with a fixed ``seed`` — will generally
produce a *different* shuffle order each time, since each call
consumes further random state rather than resetting it.

Parameters:

Name	Type	Description	Default
seed	int | None	Seed for the internal random number generator, to make the shuffle order reproducible across runs. If None, the generator is seeded unpredictably (e.g. from system entropy).	None

Example:

>>> from zenpyre.data_processors import ShuffleProcessor
>>> processor = ShuffleProcessor(seed=42)
>>> shuffled = processor.process([1, 2, 3, 4, 5])
>>> sorted(shuffled)
[1, 2, 3, 4, 5]

zenpyre.data_processors.ShuffleProcessor.process

process(data: Sequence[T]) -> list[T]

Shuffle the items of the input sequence.

Converts the input to a list and shuffles it in place using the internal random generator.

Parameters:

Name	Type	Description	Default
data	Sequence[T]	The sequence to shuffle. Can be any iterable (e.g. list, tuple).	required

Returns:

Type	Description
list[T]	A new list containing the same elements as data,
list[T]	in a random order.

Raises:

Type	Description
TypeError	If data is not iterable.

zenpyre.data_processors.SortByKeyProcessor

Bases: BaseProcessor[Sequence[dict[str, Any]], list[dict[str, Any]]], InlineDisplayMixin

Processor that sorts a sequence of dicts by the value at a given key.

Parameters:

Name	Type	Description	Default
key	str	The dict key whose value is used as the sort key. Every dict in the input must contain this key, and the values must be mutually comparable (support <).	required
reverse	bool	If True, sort in descending order. Defaults to False (ascending order), matching the built-in :func:sorted function.	False

Raises:

Type	Description
KeyError	If key is missing from a dict in the input.

Example

>>> from zenpyre.data_processors import SortByKeyProcessor
>>> processor = SortByKeyProcessor(key="score")
>>> processor.process([{"score": 3}, {"score": 1}, {"score": 2}])
[{'score': 1}, {'score': 2}, {'score': 3}]
>>> processor = SortByKeyProcessor(key="score", reverse=True)
>>> processor.process([{"score": 3}, {"score": 1}, {"score": 2}])
[{'score': 3}, {'score': 2}, {'score': 1}]

zenpyre.data_processors.SortByKeyProcessor.process

process(
    data: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]

Sort the dicts by the value at :attr:_key.

Parameters:

Name	Type	Description	Default
data	Sequence[dict[str, Any]]	The sequence of dicts to sort. Every dict must contain :attr:_key.	required

Returns:

Type	Description
list[dict[str, Any]]	A new list of dicts sorted by the value at :attr:_key,
list[dict[str, Any]]	in ascending order, or descending order if
list[dict[str, Any]]	reverse=True.

Raises:

Type	Description
KeyError	If :attr:_key is missing from a dict.

zenpyre.data_processors.SortDocumentsByMetadataProcessor

Bases: BaseProcessor[list[Document], list[Document]], InlineDisplayMixin

Processor that sorts a list of LangChain documents by the value of a metadata key.

Wraps :func:~zenpyre.documents.sort_by_metadata as a :class:~zenpyre.data_processors.base.BaseProcessor so it can be composed in a :class:~zenpyre.data_processors.SequentialProcessor pipeline.

Parameters:

Name	Type	Description	Default
metadata_key	str	The metadata key to sort by.	required
keep_missing	bool	If True (the default), documents without metadata_key are kept and placed at the end of the result. If False, they are excluded entirely.	True
reverse	bool	If True, the result is sorted in descending order. Defaults to False.	False

Example

>>> from langchain_core.documents import Document
>>> from zenpyre.data_processors import SortDocumentsByMetadataProcessor
>>> processor = SortDocumentsByMetadataProcessor(metadata_key="source")
>>> docs = [
...     Document(page_content="B", metadata={"source": "b.txt"}),
...     Document(page_content="A", metadata={"source": "a.txt"}),
... ]
>>> result = processor.process(docs)
>>> [doc.metadata["source"] for doc in result]
['a.txt', 'b.txt']

zenpyre.data_processors.SortDocumentsByMetadataProcessor.process

process(data: list[Document]) -> list[Document]

Sort data by the configured metadata key and return the result.

Parameters:

Name	Type	Description	Default
data	list[Document]	The list of :class:~langchain_core.documents.Document instances to sort.	required

Returns:

Type	Description
list[Document]	A new sorted list of
list[Document]	class:~langchain_core.documents.Document instances.
list[Document]	The original list is not modified.

zenpyre.data_processors.SortRecordsByMetadataProcessor

Bases: BaseProcessor[list[Record], list[Record]], InlineDisplayMixin

Processor that sorts a list of records by the value of a metadata key.

Wraps :func:~sort_by_metadata_record.sort_by_metadata as a :class:~zenpyre.data_processors.base.BaseProcessor so it can be composed in a :class:~zenpyre.data_processors.SequentialProcessor pipeline.

Parameters:

Name	Type	Description	Default
metadata_key	str	The metadata key to sort by.	required
keep_missing	bool	If True (the default), records without metadata_key are kept and placed at the end of the result. If False, they are excluded entirely.	True
reverse	bool	If True, the result is sorted in descending order. Defaults to False.	False

Example

>>> from zenpyre.records import Record
>>> from zenpyre.data_processors import SortRecordsByMetadataProcessor
>>> processor = SortRecordsByMetadataProcessor(metadata_key="source")
>>> records = [
...     Record(id="b", metadata={"source": "b.txt"}),
...     Record(id="a", metadata={"source": "a.txt"}),
... ]
>>> result = processor.process(records)
>>> [r.metadata["source"] for r in result]
['a.txt', 'b.txt']

zenpyre.data_processors.SortRecordsByMetadataProcessor.process

process(data: list[Record]) -> list[Record]

Sort data by the configured metadata key and return the result.

Parameters:

Name	Type	Description	Default
data	list[Record]	The list of :class:~zenpyre.records.Record instances to sort.	required

Returns:

Type	Description
list[Record]	A new sorted list of :class:~zenpyre.records.Record instances.
list[Record]	The original list is not modified.

zenpyre.data_processors.resolve_data_processor

resolve_data_processor(
    processor: (
        BaseProcessor[U, T]
        | dict[str, Any]
        | BaseConfig
    ),
) -> BaseProcessor[U, T]

Resolve a :class:~zenpyre.data_processors.base.BaseProcessor instance from an existing object, a configuration dictionary, or a :class:~zenpyre.utils.config.BaseConfig.

If processor is already a :class:~zenpyre.data_processors.base.BaseProcessor instance it is returned as-is. If it is a :class:dict or a :class:~zenpyre.utils.config.BaseConfig, it is treated as an objectory factory configuration and instantiated via :func:objectory.factory. See :func:~zenpyre.utils.resolve.resolve_object for details.

Parameters:

Name	Type	Description	Default
processor	BaseProcessor[U, T] | dict[str, Any] | BaseConfig	Either a fully configured :class:~zenpyre.data_processors.base.BaseProcessor instance, a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name), or a :class:~zenpyre.utils.config.BaseConfig whose to_kwargs() includes a "_target_" key.	required

Returns:

Type	Description
BaseProcessor[U, T]	A configured
BaseProcessor[U, T]	class:~zenpyre.data_processors.base.BaseProcessor
BaseProcessor[U, T]	instance.

Raises:

Type	Description
TypeError	If the resolved object is not a :class:~zenpyre.data_processors.base.BaseProcessor instance.

Example

>>> from zenpyre.data_processors import resolve_data_processor, FirstNProcessor
>>> # From an existing instance:
>>> processor = resolve_data_processor(FirstNProcessor(n=5))
>>> # From a configuration dictionary:
>>> processor = resolve_data_processor(  # doctest: +SKIP
...     {"_target_": "zenpyre.data_processors.FirstNProcessor", "n": 5}
... )

zenpyre.data_processors.factory

Contain factories for data processors.

zenpyre.data_processors.factory.BaseProcessorFactory

Bases: ABC, Generic[U, T]

Abstract base class for :class:~zenpyre.data_processors.base.BaseProcessor factories.

Subclasses implement :meth:make_processor to instantiate and return a configured :class:~zenpyre.data_processors.base.BaseProcessor object. This pattern decouples processor creation from the rest of the codebase, making it easy to swap processors (e.g. filtering, sorting, sequencing) without changing call sites.

Example

>>> from zenpyre.data_processors import FirstNProcessor
>>> from zenpyre.data_processors.base import BaseProcessor
>>> from zenpyre.data_processors.factory import BaseProcessorFactory
>>> class MyProcessorFactory(BaseProcessorFactory[list[int], list[int]]):
...     def make_processor(self) -> BaseProcessor[list[int], list[int]]:
...         return FirstNProcessor(n=5)
...
>>> factory = MyProcessorFactory()
>>> processor = factory.make_processor()

zenpyre.data_processors.factory.BaseProcessorFactory.make_processor abstractmethod

make_processor() -> BaseProcessor[U, T]

Create and return a configured BaseProcessor instance.

Returns:

Name	Type	Description
A	BaseProcessor[U, T]	class:~zenpyre.data_processors.base.BaseProcessor
	BaseProcessor[U, T]	instance ready for use.

zenpyre.data_processors.factory.ConfigurableProcessorFactory

Bases: BaseProcessorFactory[U, T], MultilineDisplayMixin

A concrete BaseProcessor factory that accepts either a pre-built :class:~zenpyre.data_processors.base.BaseProcessor instance or a configuration dictionary.

When a dict is provided it is resolved at each :meth:make_processor call via :func:~zenpyre.data_processors.resolve.resolve_data_processor, which uses objectory to instantiate the configured class. When an instance is provided it is returned as-is.

Parameters:

Name	Type	Description	Default
processor	BaseProcessor[U, T] | dict[str, Any]	A fully configured :class:~zenpyre.data_processors.base.BaseProcessor instance, or a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name).	required

Example

>>> from zenpyre.data_processors import FirstNProcessor
>>> from zenpyre.data_processors.factory import ConfigurableProcessorFactory
>>> factory = ConfigurableProcessorFactory(FirstNProcessor(n=5))
>>> processor = factory.make_processor()

zenpyre.data_processors.factory.ProcessorFactory

Bases: BaseProcessorFactory[U, T], MultilineDisplayMixin

A concrete BaseProcessor factory that wraps a pre-built :class:~zenpyre.data_processors.base.BaseProcessor instance.

Use this when the processor is already instantiated and you simply want to wrap it in the :class:~BaseProcessorFactory interface — for example, when injecting a fixed processor into a component that expects a factory.

Parameters:

Name	Type	Description	Default
processor	BaseProcessor[U, T]	A fully configured :class:~zenpyre.data_processors.base.BaseProcessor instance to return from :meth:make_processor.	required

Example

>>> from zenpyre.data_processors import FirstNProcessor
>>> from zenpyre.data_processors.factory import ProcessorFactory
>>> factory = ProcessorFactory(FirstNProcessor(n=5))
>>> processor = factory.make_processor()