Ingestors

zenpyre.ingestors

Contain all ingestors.

zenpyre.ingestors.BaseIngestor

Bases: ABC, Generic[T]

Abstract base class for data ingestors.

A generic base for any component that ingests data — reads it from a source, transforms it, writes it to a store, or any combination thereof. Works with any payload type T, including :class:~langchain_core.documents.Document objects, plain strings, dataclasses, or None (for ingestors that write to a store and return nothing).

The contract intentionally leaves the source, transport, and schema to the concrete implementation. Subclasses must implement :meth:ingest.

Example

>>> from zenpyre.ingestors import BaseIngestor
>>> from langchain_core.documents import Document
>>> class MyDocumentIngestor(BaseIngestor[list[Document]]):
...     def ingest(self) -> list[Document]:
...         return [Document(page_content="Hello")]
...
>>> ingestor = MyDocumentIngestor()
>>> docs = ingestor.ingest()

zenpyre.ingestors.BaseIngestor.ingest abstractmethod

ingest() -> T

Ingest data from the configured source.

Depending on the implementation, this may read from a file, database, API, or any other source, optionally transform the data, and return it or write it to a destination store.

Returns:

Type	Description
T	The ingested payload. Its type and structure are defined
T	by the concrete implementation. May be None for
T	ingestors that write to a store as a side effect.

zenpyre.ingestors.DataclassIngestor

Bases: BaseIngestor[list[T]], InlineDisplayMixin, Generic[T]

An ingestor that loads a list of dataclass instances from a JSON file.

Reads a JSON file previously written by :func:~zenpyre.utils.dataclass_io.save_dataclasses and reconstructs each entry as an instance of cls.

Parameters:

Name	Type	Description	Default
path	Path | str	The path to the JSON file to load.	required
cls	type[T]	The dataclass type to reconstruct each entry as.	required

Example

>>> from dataclasses import dataclass
>>> from zenpyre.ingestors import DataclassIngestor
>>> @dataclass(frozen=True)
... class Point:
...     x: int
...     y: int
...
>>> ingestor = DataclassIngestor(path="points.json", cls=Point)
>>> points = ingestor.ingest()  # doctest: +SKIP

zenpyre.ingestors.FirstNIngestor

Bases: BaseIngestor[list[T]], MultilineDisplayMixin

Ingestor that returns the first n items from another ingestor.

Wraps a source :class:~zenpyre.ingestors.BaseIngestor that returns a sequence, and slices the first n elements from its output. If the source ingestor returns fewer than n items, all items are returned.

Parameters:

Name	Type	Description	Default
source	BaseIngestor[Sequence[T]]	The source ingestor whose output will be sliced. Must return a sequence.	required
n	int	The maximum number of items to return from the start of the sequence. Must be a positive integer.	required

Raises:

Type	Description
ValueError	If n is not a positive integer.

Example

>>> from zenpyre.ingestors import InMemoryIngestor, FirstNIngestor
>>> ingestor = FirstNIngestor(
...     source=InMemoryIngestor(data=[1, 2, 3, 4, 5]),
...     n=3,
... )
>>> ingestor.ingest()
[1, 2, 3]

zenpyre.ingestors.FirstNIngestor.ingest

ingest() -> list[T]

Return the first n items from the source ingestor.

Returns:

Type	Description
list[T]	A list containing the first n elements of the source
list[T]	ingestor's output. If fewer than n items are available,
list[T]	all of them are returned.

zenpyre.ingestors.InMemoryIngestor

Bases: BaseIngestor[T], InlineDisplayMixin

Ingestor that returns a pre-loaded in-memory value.

Wraps an arbitrary value and exposes it through the :meth:ingest interface, allowing any data already in memory to be used wherever a :class:~zenpyre.ingestors.BaseIngestor is expected. Useful for testing, rapid prototyping, or bypassing the download and caching steps of a pipeline.

By default each call to :meth:ingest returns a deep copy of the stored value, preventing callers from accidentally mutating the ingestor's internal state. Pass copy=False to return the original object directly, which is useful in tests where mock identity must be preserved.

Parameters:

Name	Type	Description	Default
data	T	The value to return when :meth:ingest is called.	required
copy	bool	If True (the default), return a deep copy of data on each :meth:ingest call. If False, return the original object.	True

Example

>>> from zenpyre.ingestors import InMemoryIngestor
>>> ingestor = InMemoryIngestor(data="hello\nworld")
>>> ingestor.ingest()
hello\nworld

zenpyre.ingestors.LastNIngestor

Bases: BaseIngestor[list[T]], MultilineDisplayMixin

Ingestor that returns the last n items from another ingestor.

Wraps a source :class:~zenpyre.ingestors.BaseIngestor that returns a sequence, and slices the last n elements from its output. If the source ingestor returns fewer than n items, all items are returned.

Parameters:

Name	Type	Description	Default
ingestor		The source ingestor whose output will be sliced. Must return a sequence.	required
n	int	The maximum number of items to return from the end of the sequence. Must be a positive integer.	required

Raises:

Type	Description
ValueError	If n is not a positive integer.

Example

>>> from zenpyre.ingestors import InMemoryIngestor
>>> ingestor = LastNIngestor(
...     source=InMemoryIngestor(data=[1, 2, 3, 4, 5]),
...     n=3,
... )
>>> ingestor.ingest()
[3, 4, 5]

zenpyre.ingestors.LastNIngestor.ingest

ingest() -> list[T]

Return the last n items from the source ingestor.

Returns:

Type	Description
list[T]	A list containing the last n elements of the source
list[T]	ingestor's output. If fewer than n items are available,
list[T]	all of them are returned.

zenpyre.ingestors.MappingIngestor

Bases: BaseIngestor[dict[Hashable, BaseIngestor[T]]], MultilineDisplayMixin

Ingestor that calls a mapping of ingestors and returns their results as a dict.

Each inner ingestor is called in order and its return value is stored under the corresponding key. The types of the individual results are unconstrained — each ingestor may return a different type.

Parameters:

Name	Type	Description	Default
sources	Mapping[Hashable, BaseIngestor[T]]	A mapping from arbitrary keys to :class:~zenpyre.ingestors.BaseIngestor instances.	required

Example

>>> from zenpyre.ingestors import InMemoryIngestor, MappingIngestor
>>> ingestor = MappingIngestor(
...     sources={
...         "10-K": InMemoryIngestor(data="annual report text"),
...         "10-Q": InMemoryIngestor(data="quarterly report text"),
...     }
... )
>>> ingestor.ingest()
{'10-K': 'annual report text', '10-Q': 'quarterly report text'}

zenpyre.ingestors.PickleIngestor

Bases: BaseIngestor[T], InlineDisplayMixin

Ingestor that loads data from a pickle file.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the pickle file to load.	required

Raises:

Type	Description
FileNotFoundError	If path does not exist when :meth:ingest is called.

Example

>>> ingestor = PickleIngestor(path="data.pkl")
>>> data = ingestor.ingest()  # doctest: +SKIP

zenpyre.ingestors.PickleIngestor.ingest

ingest() -> T

Load and return data from the pickle file.

Returns:

Type	Description
T	The Python object stored in the pickle file.

Raises:

Type	Description
FileNotFoundError	If the pickle file does not exist at the configured path.

zenpyre.ingestors.ProcessorIngestor

Bases: BaseIngestor[T], MultilineDisplayMixin

Ingestor that applies a processor to the output of another ingestor.

Wraps a source :class:`~zenpyre.ingestors.base.BaseIngestor` and a
:class:`~zenpyre.data_processors.base.BaseProcessor`: it ingests
data from the source, then passes it through the processor and
returns the processed result. This makes it possible to compose
any processor (e.g. :class:`~zenpyre.data_processors.ShuffleProcessor`)
with any ingestor without needing a dedicated ingestor subclass for
each processor.

Type parameters:
    U: The type of data returned by the source ingestor, and
        accepted as input by the processor.
    T: The type of data returned by the processor, and returned by
        :meth:`ingest`.

Parameters:

Name	Type	Description	Default
source	BaseIngestor[U]	The ingestor used to fetch the raw data.	required
processor	BaseProcessor[U, T]	The processor used to transform the data ingested from source.	required

Example:

>>> from zenpyre.data_processors import ShuffleProcessor
>>> from zenpyre.ingestors import InMemoryIngestor, ProcessorIngestor
>>> ingestor = ProcessorIngestor(
...     source=InMemoryIngestor(data=[1, 2, 3, 4, 5]),
...     processor=ShuffleProcessor(seed=42),
... )
>>> sorted(ingestor.ingest())
[1, 2, 3, 4, 5]

zenpyre.ingestors.ProcessorIngestor.ingest

ingest() -> T

Ingest data from the source ingestor and process it.

Calls :meth:~zenpyre.ingestors.base.BaseIngestor.ingest on the source ingestor, then passes the result to :meth:~zenpyre.data_processors.base.BaseProcessor.process.

Returns:

Type	Description
T	The processed output, as returned by the processor.

zenpyre.ingestors.ShuffleIngestor

Bases: BaseIngestor[list[T]], MultilineDisplayMixin

Ingestor that shuffles the output of another ingestor.

Wraps a source :class:~zenpyre.ingestors.base.BaseIngestor and returns its ingested payload as a list with elements shuffled in random order. The source ingestor's output must be an iterable (e.g. list, tuple) whose elements can be reordered; the result is always materialized as a new list, so the original sequence type of T 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:ingest. This means calling ingest 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
source	BaseIngestor[Sequence[T]]	The ingestor whose output will be shuffled.	required
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.ingestors import ShuffleIngestor, InMemoryIngestor
>>> ingestor = ShuffleIngestor(InMemoryIngestor([1, 2, 3, 4, 5]), seed=42)
>>> shuffled = ingestor.ingest()
>>> sorted(shuffled)
[1, 2, 3, 4, 5]

zenpyre.ingestors.ShuffleIngestor.ingest

ingest() -> list[T]

Ingest data from the source ingestor and shuffle it.

Calls :meth:~zenpyre.ingestors.base.BaseIngestor.ingest on the wrapped source ingestor, converts the result to a list, and shuffles it in place using the internal random generator.

Returns:

Type	Description
list[T]	A new list containing the same elements as the
list[T]	source ingestor's output, in a random order.

Raises:

Type	Description
TypeError	If the source ingestor's output is not iterable.

zenpyre.ingestors.TextIngestor

Bases: BaseIngestor[str], InlineDisplayMixin

Ingestor that loads data from a text file.

Reads the full content of a text file and returns it as a single string. Intended for loading Markdown filings or other plain-text documents for downstream processing by an agent.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the text file to load.	required
encoding	str	The file encoding to use when reading the file. Defaults to "utf-8".	'utf-8'

Raises:

Type	Description
FileNotFoundError	If path does not exist when :meth:ingest is called.

Example

>>> from zenpyre.ingestors import TextIngestor
>>> ingestor = TextIngestor(path="filing.md")
>>> text = ingestor.ingest()  # doctest: +SKIP

zenpyre.ingestors.TextIngestor.ingest

ingest() -> str

Load and return the content of the text file.

Returns:

Type	Description
str	The full file content as a string.

Raises:

Type	Description
FileNotFoundError	If the text file does not exist at the configured path.

zenpyre.ingestors.resolve_ingestor

resolve_ingestor(
    ingestor: (
        BaseIngestor[T] | dict[str, Any] | BaseConfig
    ),
) -> BaseIngestor[T]

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

If ingestor is already a :class:~zenpyre.ingestors.base.BaseIngestor 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
ingestor	BaseIngestor[T] | dict[str, Any] | BaseConfig	Either a fully configured :class:~zenpyre.ingestors.base.BaseIngestor 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
BaseIngestor[T]	A configured :class:~zenpyre.ingestors.base.BaseIngestor
BaseIngestor[T]	instance.

Raises:

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

Example

>>> from zenpyre.ingestors import resolve_ingestor
>>> from zenpyre.ingestors.base import BaseIngestor
>>> class MyIngestor(BaseIngestor):
...     def ingest(self) -> Any:
...         return {"hello": "world"}
...
>>> # From an existing instance:
>>> ingestor = resolve_ingestor(MyIngestor())
>>> # From a configuration dictionary:
>>> ingestor = resolve_ingestor(  # doctest: +SKIP
...     {"_target_": "my_package.ingestors.MyIngestor"}
... )

zenpyre.ingestors.factory

Contain factories for document ingestors.

zenpyre.ingestors.factory.BaseIngestorFactory

Bases: ABC, Generic[T]

Abstract base class for :class:~zenpyre.ingestors.base.BaseIngestor factories.

Subclasses implement :meth:make_ingestor to instantiate and return a configured :class:~zenpyre.ingestors.base.BaseIngestor object. This pattern decouples ingestor creation from the rest of the codebase, making it easy to swap ingestors (e.g. file, web, database) without changing call sites.

Example

>>> from zenpyre.ingestors import InMemoryIngestor
>>> from zenpyre.ingestors.base import BaseIngestor
>>> from zenpyre.ingestors.factory import BaseIngestorFactory
>>> class MyIngestorFactory(BaseIngestorFactory[list[int]]):
...     def make_ingestor(self) -> BaseIngestor[list[int]]:
...         return InMemoryIngestor([1, 2, 3])
...
>>> factory = MyIngestorFactory()
>>> ingestor = factory.make_ingestor()

zenpyre.ingestors.factory.BaseIngestorFactory.make_ingestor abstractmethod

make_ingestor() -> BaseIngestor[T]

Create and return a configured BaseIngestor instance.

Returns:

Name	Type	Description
A	BaseIngestor[T]	class:~zenpyre.ingestors.base.BaseIngestor
	BaseIngestor[T]	instance ready for use.

zenpyre.ingestors.factory.ConfigurableIngestorFactory

Bases: BaseIngestorFactory[T], MultilineDisplayMixin

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

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

Parameters:

Name	Type	Description	Default
ingestor	BaseIngestor[T] | dict[str, Any]	A fully configured :class:~zenpyre.ingestors.base.BaseIngestor 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.ingestors import InMemoryIngestor
>>> from zenpyre.ingestors.factory import ConfigurableIngestorFactory
>>> factory = ConfigurableIngestorFactory(InMemoryIngestor([1, 2, 3]))
>>> ingestor = factory.make_ingestor()

zenpyre.ingestors.factory.IngestorFactory

Bases: BaseIngestorFactory[T], MultilineDisplayMixin

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

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

Parameters:

Name	Type	Description	Default
ingestor	BaseIngestor[T]	A fully configured :class:~zenpyre.ingestors.base.BaseIngestor instance to return from :meth:make_ingestor.	required

Example

>>> from zenpyre.ingestors import InMemoryIngestor
>>> from zenpyre.ingestors.factory import IngestorFactory
>>> factory = IngestorFactory(InMemoryIngestor([1, 2, 3]))
>>> ingestor = factory.make_ingestor()