API Reference¶

This reference is generated from the code and its docstrings. Keep public classes and methods documented as their contracts evolve.

Core contracts and data model¶

`etlrelay.core.batch.Batch` ¶

Represent a batch of tabular data using a PyArrow Table.

Batch provides a consistent data representation for sources, processors, and sinks throughout a ETLRelay pipeline.

Parameters:

Name	Type	Description	Default
`table`	`Table`	The PyArrow Table containing the batch's data.	required

`table` `property` ¶

Return the PyArrow Table containing the batch's data.

Returns:

Type	Description
`Table`	The underlying PyArrow Table.

`etlrelay.core.source.Source` ¶

Bases: Protocol

Define the interface for components that produce data batches.

Concrete source implementations are responsible for obtaining data from their respective data sources and exposing it as an iterable of Batch objects.

The protocol is runtime-checkable, allowing implementations to be checked for protocol conformance using isinstance().

`read()` ¶

Return an iterable of batches obtained from the data source.

Concrete implementations may raise implementation-specific exceptions when the underlying data cannot be accessed or read.

Returns:

Type	Description
`Iterable[Batch]`	An iterable of Batch objects produced by the source.

`etlrelay.core.processor.Processor` ¶

Bases: Protocol

Define the interface for ETLRelay batch processors.

A processor accepts a Batch, applies a transformation, and returns the resulting Batch. Processors are executed sequentially by the pipeline.

The protocol allows different processor implementations to be used interchangeably without coupling the pipeline to concrete classes.

`process(batch)` ¶

Transform a batch and return the resulting batch.

Parameters:

Name	Type	Description	Default
`batch`	`Batch`	The input Batch to process.	required

Returns:

Type	Description
`Batch`	The resulting Batch after applying the transformation.

`etlrelay.core.sink.Sink` ¶

Bases: Protocol

Define the interface for components that consume batches.

Concrete implementations are responsible for delivering batches to their destination, such as a file or database. The protocol is independent of any particular storage backend or output format.

`write(batches)` ¶

Write an iterable of batches to the configured destination.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[Batch]`	An iterable of batches to consume and write.	required

Raises:

Type	Description
`Exception`	If writing fails. Concrete implementations may raise more specific exceptions for their respective failure cases.

`etlrelay.core.reader.Reader` ¶

Bases: Protocol

Protocol for deserializing binary streams into ETLRelay batches.

Implementations handle format-specific parsing independently of the storage system that provides the stream.

`read(stream)` ¶

Read and deserialize data from a binary stream.

Parameters:

Name	Type	Description	Default
`stream`	`BinaryIO`	An open binary stream containing serialized input data.	required

Returns:

Type	Description
`Iterable[Batch]`	An iterable of batches containing the parsed data.

Raises:

Type	Description
`ReaderError`	If the input data cannot be read or parsed.

Note

The caller is responsible for managing the stream's lifecycle.

`etlrelay.core.writer.Writer` ¶

Bases: Protocol

Define the interface for serializing batches to a binary stream.

Implementations are responsible for format-specific serialization. They do not manage filesystem access or the lifecycle of the supplied stream.

`write(batches, stream)` ¶

Serialize batches and write the resulting data to a stream.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[Batch]`	An iterable of batches to serialize into one output.	required
`stream`	`BinaryIO`	A writable binary stream that receives the serialized data. The caller is responsible for closing the stream.	required

Raises:

Type	Description
`WriterError`	If the batches cannot be serialized.
`OSError`	If an I/O error occurs while writing to the stream.

Pipeline¶

`etlrelay.pipeline.pipeline.Pipeline` ¶

Coordinate batch processing between a source, processors, and a sink.

The pipeline obtains batches from a source, applies each configured processor in sequence, validates the resulting output schemas, and passes compatible batches to the sink.

The first processed output batch establishes the expected schema for the run. Subsequent output batches must have matching field names, field order, Arrow types, and nullability. Schema metadata differences are ignored. Zero-row batches participate in schema validation.

Processing is lazy and does not buffer all batches before writing. If a later batch has an incompatible schema, a ValueError is raised before that batch is yielded to the sink. Batches successfully written before a later failure are not rolled back.

Exceptions from the source, processors, or sink otherwise propagate to the caller.

`init(source, processors, sink)` ¶

Initialize the pipeline with its source, processors, and sink.

Parameters:

Name	Type	Description	Default
`source`	`Source`	Component that produces batches for processing.	required
`processors`	`Sequence[Processor]`	Ordered sequence of processors to apply to each batch. May be empty if no transformations are required.	required
`sink`	`Sink`	Component that consumes the processed batches.	required

`run()` ¶

Execute the pipeline and deliver processed batches to the sink.

Batches are obtained from the source, processed in sequence, and passed to the sink as an iterable. Processing is lazy, avoiding buffering all batches in memory before writing. If the source produces no batches, the sink is not called.

Raises:

Type	Description
`ValueError`	If a processed batch has a schema incompatible with the first processed output batch.
`Exception`	Exceptions raised during source iteration, processing, or sink execution propagate to the caller.

Previously written batches are not rolled back if a later batch fails.

Processors¶

`etlrelay.processors.select_columns.SelectColumns` ¶

Select specific columns from a batch in the requested order.

The processor returns a new Batch containing only the selected columns. Column values, Arrow data types, and nullability are preserved.

Duplicate requested column names and unknown input columns raise ValueError.

Example

from etlrelay.processors import SelectColumns

processor = SelectColumns(columns=["name", "age"]) result = processor.process(batch)

`init(columns)` ¶

Initialize the processor with the columns to select.

Parameters:

Name	Type	Description	Default
`columns`	`list[str]`	Names of the columns to select, in the desired output order.	required

Raises:

Type	Description
`ValueError`	If duplicate column names are provided.

`process(batch)` ¶

Select the configured columns from a batch.

Parameters:

Name	Type	Description	Default
`batch`	`Batch`	The input batch containing the columns to select.	required

Returns:

Type	Description
`Batch`	A new Batch containing only the requested columns in the
`Batch`	specified order.

Raises:

Type	Description
`ValueError`	If any requested column does not exist in the input batch.

`etlrelay.processors.rename_columns.RenameColumns` ¶

`init(mapping)` ¶

Rename columns using a mapping of existing names to new names.

Column order, values, Arrow data types, and nullability are preserved. Multiple columns can be renamed simultaneously, including swapping names, provided that the final column names are unique.

Validation occurs when processing a batch. Unknown source columns, duplicate destination names, and conflicts with existing columns that are not being renamed raise ValueError.

Example

from etlrelay.processors import RenameColumns

processor = RenameColumns(mapping={"name": "customer_name"}) result = processor.process(batch)

`process(batch)` ¶

Rename columns in a batch.

Parameters:

Name	Type	Description	Default
`batch`	`Batch`	The batch whose columns will be renamed.	required

Returns:

Type	Description
`Batch`	A new Batch with renamed columns, preserving column order, values,
`Batch`	data types, and nullability.

Raises:

Type	Description
`ValueError`	If a source column does not exist or the mapping would produce duplicate column names.

`etlrelay.processors.filter_rows.FilterRows` ¶

Filter a batch using a single column comparison predicate.

Supported operators are ==, !=, <, <=, >, and >=. Comparisons use Apache Arrow compute operations without custom casting.

Rows whose comparison evaluates to false or null are excluded. The input schema is preserved, including when no rows match.

The processor can be reused across batches without retaining batch-specific state.

Example

from etlrelay.processors import FilterRows

processor = FilterRows(column="age", operator=">=", value=18) result = processor.process(batch)

Raises:

Type	Description
`ValueError`	If the comparison operator is unsupported.

`init(column, operator, value)` ¶

Initialize the processor with a comparison predicate.

Parameters:

Name	Type	Description	Default
`column`	`str`	Name of the column to compare.	required
`operator`	`str`	Comparison operator. Must be one of ==, !=, <, <=, , or >=.	required
`value`	`Any`	Value to compare against the selected column.	required

Raises:

Type	Description
`ValueError`	If the operator is unsupported.

`process(batch)` ¶

Return a batch containing only rows that satisfy the predicate.

Rows whose comparison evaluates to false or null are excluded. The resulting batch preserves the input table's schema.

Parameters:

Name	Type	Description	Default
`batch`	`Batch`	Batch containing the column to filter.	required

Returns:

Type	Description
`Batch`	A new Batch containing the matching rows, with the original
`Batch`	schema preserved.

Raises:

Type	Description
`ValueError`	If the specified column does not exist or is ambiguous.
`TypeError`	If the column and comparison value have incompatible types.

Filesystem and security¶

`etlrelay.security.path.PathValidator` ¶

Validate filesystem paths against a trusted base directory.

PathValidator resolves both the configured base directory and candidate paths before checking whether the candidate remains within the permitted directory. This prevents path traversal and symlink-based paths from escaping the configured boundary.

Parameters:

Name	Type	Description	Default
`base_path`	`Path \| None`	The directory that candidate paths must remain within. If None, the current working directory is used.	`None`

Raises:

Type	Description
`PathValidationError`	Raised by validate() when a candidate path is outside the base directory or does not exist.

`validate(path)` ¶

Validate and resolve a filesystem path.

The candidate path is resolved relative to the configured base directory. The resolved path must remain within the base directory and must exist on the filesystem.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The filesystem path to validate. May be a Path object or a string. Relative paths are resolved relative to the configured base directory.	required

Returns:

Type	Description
`Path`	The resolved Path for the validated filesystem location.

Raises:

Type	Description
`PathValidationError`	If the resolved path is outside the configured base directory or does not exist.

`etlrelay.storage.filesystem.FileSystem` ¶

Bases: Protocol

Define the interface for accessing file resources.

Implementations provide readable and writable binary streams for storage resources. Callers are responsible for closing the returned streams, preferably by using a context manager.

`open_read(path)` ¶

Open a resource for binary reading.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path or resource identifier understood by the filesystem implementation.	required

Returns:

Type	Description
`BinaryIO`	A readable binary stream.

Raises:

Type	Description
`OSError`	If the resource cannot be opened for reading.

`open_write(path)` ¶

Open a resource for binary writing.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path or resource identifier understood by the filesystem implementation.	required

Returns:

Type	Description
`BinaryIO`	A writable binary stream.

Raises:

Type	Description
`OSError`	If the resource cannot be opened for writing.

`etlrelay.storage.local.LocalFileSystem` ¶

Provide validated binary I/O for the local filesystem.

LocalFileSystem uses PathValidator to enforce the permitted filesystem boundary before opening files. It returns binary streams so that format-specific readers and writers can operate independently of the underlying storage implementation.

Parameters:

Name	Type	Description	Default
`path_validator`	`PathValidator`	Validator used to ensure paths remain within the permitted base directory.	required

`open_read(path)` ¶

Open a validated local file for binary reading.

The path is validated and must refer to an existing filesystem resource that can be opened for reading.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to the file, relative to the validator's base directory or an absolute path permitted by the validator.	required

Returns:

Type	Description
`BinaryIO`	A readable binary stream. The caller is responsible for closing
`BinaryIO`	the stream, preferably by using a context manager.

Raises:

Type	Description
`PathValidationError`	If the path resolves outside the permitted base directory.
`FileSystemError`	If the path does not exist.
`OSError`	If the file cannot be opened or read.

`open_write(path)` ¶

Open a validated local file for binary writing.

The path is validated before the file is opened. If the destination does not exist, it is created. If it already exists, its contents are truncated. Parent directories are not created automatically.

Parameters:

Name	Type	Description	Default
`path`	`str`	Destination path, relative to the validator's base directory or an absolute path permitted by the validator.	required

Returns:

Type	Description
`BinaryIO`	A writable binary stream. The caller is responsible for closing
`BinaryIO`	the stream, preferably by using a context manager.

Raises:

Type	Description
`PathValidationError`	If the path resolves outside the permitted base directory.
`OSError`	If the destination cannot be opened for writing, such as when its parent directory does not exist or access is denied.

File source and sink¶

`etlrelay.sources.file.FileSource` ¶

Read data from a file and expose it as an iterable of batches.

Attributes:

Name	Type	Description
`path`	`str`	The path to the file to read.
`storage`	`FileSystem`	The filesystem used to open the file.
`reader`	`Reader`	The reader used to parse the file contents into a batch.

`init(path, storage, reader)` ¶

Initialize the file source.

Parameters:

Name	Type	Description	Default
`path`	`str`	The path to the file to read.	required
`storage`	`FileSystem`	The filesystem used to access the file.	required
`reader`	`Reader`	The reader used to parse the file contents.	required

`read()` ¶

Read the file and return its data as an iterable of batches.

Opens the file through the configured filesystem, parses its contents using the configured reader, and returns the resulting batch in an iterable.

Returns:

Type	Description
`Iterable[Batch]`	An iterable containing the batch parsed from the file.

`etlrelay.sinks.file.FileSink` ¶

Write batches to a file resource using a storage backend and writer.

FileSink obtains a writable binary stream from the supplied FileSystem and passes the batches to the supplied Writer for serialization. It is independent of the underlying storage implementation and file format.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path identifying the output resource.	required
`storage`	`FileSystem`	Storage implementation used to open the output stream.	required
`writer`	`Writer`	Writer used to serialize the batches.	required

`init(path, storage, writer)` ¶

Initialize the file sink with its destination and dependencies.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path identifying the output resource.	required
`storage`	`FileSystem`	Storage implementation used to access the resource.	required
`writer`	`Writer`	Writer used to serialize batches.	required

`write(batches)` ¶

Serialize batches and write them to the configured destination.

Opens the destination through the storage implementation and passes the resulting binary stream and batches to the writer. The stream is closed when writing completes or an exception occurs.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[Batch]`	Iterable of batches to serialize.	required

Raises:

Type	Description
`OSError`	If the storage implementation cannot open or write to the destination.
`Exception`	Any exception raised by the writer is propagated.

CSV reader and writer¶

`etlrelay.readers.csv.CsvReader` ¶

Read CSV data from a binary stream and yield batches.

`read(stream)` ¶

Read CSV data from a binary stream and yield batches.

Parameters:

Name	Type	Description	Default
`stream`	`BinaryIO`	A binary stream containing CSV data.	required

Yields:

Type	Description
`Iterable[Batch]`	Batches containing parsed CSV data, one per record batch
`Iterable[Batch]`	produced by PyArrow's streaming CSV reader.

Raises:

Type	Description
`ArrowInvalid`	If the CSV data cannot be parsed.
`OSError`	If an error occurs while reading from the stream.

`etlrelay.writers.csv.CsvWriter` ¶

Serialize iterable ETLRelay batches as CSV data.

`write(batches, stream)` ¶

Serialize batches to CSV in the supplied binary stream.

Writes the CSV header only once, using the first batch's schema. Subsequent batches contribute rows without additional headers.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[Batch]`	Iterable of batches containing tabular data.	required
`stream`	`BinaryIO`	Writable binary stream that receives the CSV data.	required

Raises:

Type	Description
`OSError`	If an error occurs while writing to the stream.

Parquet reader and writer¶

`etlrelay.readers.parquet.ParquetReader` ¶

Read Parquet data from a binary stream and yield ETLRelay batches.

Preserves the Arrow schema for zero-row Parquet files by yielding an empty batch with the file's schema.

The caller is responsible for managing the stream's lifecycle.

Raises:

Type	Description
`ArrowInvalid`	If the Parquet data is invalid.
`OSError`	If an I/O error occurs while reading the stream.

`read(stream)` ¶

Read Parquet data from a binary stream and yield batches.

Parameters:

Name	Type	Description	Default
`stream`	`BinaryIO`	A seekable binary stream containing Parquet data.	required

Yields:

Type	Description
`Iterable[Batch]`	Batches containing the parsed data, one per record batch
`Iterable[Batch]`	produced by PyArrow. For zero-row files, yields one empty
`Iterable[Batch]`	batch preserving the file's schema.

Raises:

Type	Description
`ArrowInvalid`	If the Parquet data is invalid.
`OSError`	If an I/O error occurs while reading the stream.

`etlrelay.writers.parquet.ParquetWriter` ¶

Serialize ETLRelay batches as Parquet data.

`write(batches, stream)` ¶

Serialize batches into a single Parquet file.

Uses the first batch's schema to initialize the Parquet writer. Subsequent batches must have a matching schema and are written to the same Parquet file.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[Batch]`	An iterable of batches containing tabular data. At least one batch is required to define the schema.	required
`stream`	`BinaryIO`	A writable binary stream that receives the Parquet data. The caller is responsible for managing the stream's lifecycle.	required

Raises:

Type	Description
`WriterError`	If the iterable contains no batches or a batch has a schema incompatible with the first batch.
`OSError`	If an I/O error occurs while writing to the stream.

API Reference¶

Core contracts and data model¶

etlrelay.core.batch.Batch ¶

table property ¶

etlrelay.core.source.Source ¶

read() ¶

etlrelay.core.processor.Processor ¶

process(batch) ¶

etlrelay.core.sink.Sink ¶

write(batches) ¶

etlrelay.core.reader.Reader ¶

read(stream) ¶

etlrelay.core.writer.Writer ¶

write(batches, stream) ¶

Pipeline¶

etlrelay.pipeline.pipeline.Pipeline ¶

__init__(source, processors, sink) ¶

run() ¶

Processors¶

etlrelay.processors.select_columns.SelectColumns ¶

__init__(columns) ¶

process(batch) ¶

etlrelay.processors.rename_columns.RenameColumns ¶

__init__(mapping) ¶

process(batch) ¶

etlrelay.processors.filter_rows.FilterRows ¶

__init__(column, operator, value) ¶

process(batch) ¶

Filesystem and security¶

etlrelay.security.path.PathValidator ¶

validate(path) ¶

etlrelay.storage.filesystem.FileSystem ¶

open_read(path) ¶

open_write(path) ¶

etlrelay.storage.local.LocalFileSystem ¶

open_read(path) ¶

open_write(path) ¶

File source and sink¶

etlrelay.sources.file.FileSource ¶

__init__(path, storage, reader) ¶

read() ¶

etlrelay.sinks.file.FileSink ¶

__init__(path, storage, writer) ¶

write(batches) ¶

CSV reader and writer¶

etlrelay.readers.csv.CsvReader ¶

read(stream) ¶

etlrelay.writers.csv.CsvWriter ¶

write(batches, stream) ¶

Parquet reader and writer¶

etlrelay.readers.parquet.ParquetReader ¶

read(stream) ¶

etlrelay.writers.parquet.ParquetWriter ¶

write(batches, stream) ¶

`etlrelay.core.batch.Batch` ¶

`table` `property` ¶

`etlrelay.core.source.Source` ¶

`read()` ¶

`etlrelay.core.processor.Processor` ¶

`process(batch)` ¶

`etlrelay.core.sink.Sink` ¶

`write(batches)` ¶

`etlrelay.core.reader.Reader` ¶

`read(stream)` ¶

`etlrelay.core.writer.Writer` ¶

`write(batches, stream)` ¶

`etlrelay.pipeline.pipeline.Pipeline` ¶

`init(source, processors, sink)` ¶

`run()` ¶

`etlrelay.processors.select_columns.SelectColumns` ¶

`init(columns)` ¶

`process(batch)` ¶

`etlrelay.processors.rename_columns.RenameColumns` ¶

`init(mapping)` ¶

`process(batch)` ¶

`etlrelay.processors.filter_rows.FilterRows` ¶

`init(column, operator, value)` ¶

`process(batch)` ¶

`etlrelay.security.path.PathValidator` ¶

`validate(path)` ¶

`etlrelay.storage.filesystem.FileSystem` ¶

`open_read(path)` ¶

`open_write(path)` ¶

`etlrelay.storage.local.LocalFileSystem` ¶

`open_read(path)` ¶

`open_write(path)` ¶

`etlrelay.sources.file.FileSource` ¶

`init(path, storage, reader)` ¶

`read()` ¶

`etlrelay.sinks.file.FileSink` ¶

`init(path, storage, writer)` ¶

`write(batches)` ¶

`etlrelay.readers.csv.CsvReader` ¶

`read(stream)` ¶

`etlrelay.writers.csv.CsvWriter` ¶

`write(batches, stream)` ¶

`etlrelay.readers.parquet.ParquetReader` ¶

`read(stream)` ¶

`etlrelay.writers.parquet.ParquetWriter` ¶

`write(batches, stream)` ¶