API Reference

This reference documents ETLRelay's public Python interfaces and implementations using their source-code docstrings.

The v0.9.0 API review retains the synchronous pipeline model and configuration schema version 1. Tabular stream protocols are now named TabularReader and TabularWriter; update imports of the former class names. See the API review for decisions and compatibility notes.

Important runtime behaviors include:

• Local filesystem path containment and explicit stream ownership.
• Lazy batch processing and deterministic cleanup of closeable pipeline iterators.
• Output-schema validation, including zero-row batches.
• SQLAlchemy connection and result cleanup.
• Per-batch SQL transactions and append-only persistence.
• Configuration validation and SQL environment-variable preflight.
• Exception propagation and controlled operational logging.

The Python API generally preserves underlying runtime exceptions. The CLI provides separate error-reporting and exit-code conventions, documented in the CLI Reference.

Configuration schema version 1 remains supported and is independent of the ETLRelay package version.

The documented public interfaces and import paths are stable from v1.0.0. See the compatibility policy for guarantees, exclusions, and release/deprecation rules.

Consult the YAML Configuration, Security, Logging, and Architecture documentation for detailed usage contracts and limitations.

Configuration loading and validation

etlrelay.config.loader.ConfigLoader

Read, parse, and validate a pipeline configuration document.

The loader coordinates filesystem access, document parsing, and configuration validation. It delegates these responsibilities to injected dependencies rather than implementing filesystem operations, parsing, or validation itself.

__init__(filesystem, reader, validator)

Initialize the loader with its I/O and validation dependencies.

Parameters:

Name	Type	Description	Default
filesystem	FileSystem	Filesystem abstraction used to open the configuration document for reading.	required
reader	KeyValueReader	Reader that parses the document stream into a configuration object.	required
validator	ConfigValidator	Validator that checks the parsed configuration.	required

load(path)

Read, parse, validate, and return a pipeline configuration.

The document is opened through the configured filesystem, parsed by the configured reader, and validated before being returned. If reading or parsing fails, validation is not performed.

Parameters:

Name	Type	Description	Default
path	str	Path or resource identifier of the configuration document, interpreted by the configured filesystem.	required

Returns:

Type	Description
Mapping[str, Any]	The parsed configuration mapping after successful validation.

Raises:

Type	Description
OSError	If an underlying filesystem or I/O operation raises this exception.
Exception	Any parsing or validation exception raised by the configured reader or validator may propagate to the caller.

etlrelay.config.validator.ConfigValidator

Validate a complete versioned pipeline configuration.

The validator enforces the root configuration schema, including the supported version and required top-level fields. It delegates validation of individual pipeline sections to the injected component validators and aggregates their errors with any root-level validation errors.

Attributes:

Name	Type	Description
SUPPORTED_VERSION		The configuration schema version accepted by this validator.

__init__(source_validator, processor_validator, sink_validator)

Initialize the validator with the component-family validators.

Parameters:

Name	Type	Description	Default
source_validator	ComponentConfigValidator	Validator for the source configuration section.	required
processor_validator	ComponentConfigValidator	Validator for the ordered processor configuration sequence.	required
sink_validator	ComponentConfigValidator	Validator for the sink configuration section.	required

validate(config)

Validate the root schema and all present pipeline sections.

The root value must be a mapping with string keys. The supported top-level fields are version, source, processors, and sink, and all four fields are required. The version must be the integer specified by SUPPORTED_VERSION.

After checking the root mapping, this method delegates validation of each present pipeline section and aggregates the reported errors. Root values that are not mappings, or mappings containing non-string keys, are rejected before section validation begins.

Parameters:

Name	Type	Description	Default
config	object	Configuration value to validate.	required

Raises:

Type	Description
ConfigurationError	If the root configuration is malformed, contains unsupported or missing fields, uses an unsupported version, or contains errors reported by a section validator.

etlrelay.config.errors.ConfigurationError

Bases: Exception

Raised when a configuration-related operation fails.

This exception represents failures that prevent a configuration operation from completing successfully.

etlrelay.readers.yaml.YamlReader

Read a YAML document whose root is a string-keyed mapping.

This reader handles safe YAML parsing and enforces the root mapping contract. It does not validate application-specific schema rules or restrict the key types of nested mappings.

read(stream)

Parse a single YAML document from a binary stream.

The YAML root must be a mapping, and all root-level keys must be strings. Duplicate keys are rejected by the custom safe loader. The stream is not explicitly closed by this method.

Parameters:

Name	Type	Description	Default
stream	BinaryIO	Binary stream containing the YAML document.	required

Returns:

Type	Description
Mapping[str, Any]	The parsed root mapping, with string keys at the root. Nested
Mapping[str, Any]	mappings may contain keys of other types.

Raises:

Type	Description
YamlDocumentError	If the parsed document root is not a mapping or any root-level key is not a string.
YAMLError	If parsing fails, the document contains duplicate or unhashable mapping keys, or the stream contains multiple YAML documents instead of a single document.

etlrelay.core.reader.KeyValueReader

Bases: Protocol

Define the interface for reading one mapping-shaped document.

The result represents a document with string keys, potentially containing nested values. It is distinct from an iterable of row mappings and from the tabular batches produced by TabularReader.

Implementations parse the document independently of the storage system providing the stream.

read(stream)

Read a mapping-shaped document from a binary stream.

Parameters:

Name	Type	Description	Default
stream	BinaryIO	An open, readable binary stream containing a serialized document.	required

Returns:

Type	Description
Mapping[str, Any]	A mapping with string keys representing the parsed document.

Note

The caller owns the supplied stream. Implementations must finish reading the document before returning and must not close the stream, including when parsing fails.

Nested value types, empty-input behavior, format restrictions, and parsing errors are documented by each implementation. This protocol does not define an application configuration schema; configuration validation is a separate responsibility.

Public import policy

Use responsibility-level packages such as etlrelay.core, etlrelay.readers, etlrelay.sources, and etlrelay.pipeline. The root etlrelay package does not re-export all components. Configuration services use the module imports shown below. Protocol conformance is structural; runtime member checks do not prove stream ownership or return-value contracts.

See Migration for renamed protocol imports, Development for an executable custom-processor example, and Configuration for built-in YAML properties.

Configuration construction

etlrelay.default_components.ComponentCatalog dataclass

Bundle shared registries and services for configuration processing.

The catalog exposes the registries used to resolve storage, source, sink, and processor types, as well as the tabular file-format registry, SQL engine factory, and root configuration validator.

File sources and sinks share the catalog's storage and file-format registries. SQL sources and sinks share its SQL engine factory.

Attributes:

Name	Type	Description
tabular_file_format_registry	TabularFileFormatRegistry	Registry of supported tabular readers and writers.
storage_registry	ComponentRegistry[FileSystem]	Registry of storage component registrations.
source_registry	ComponentRegistry[Source]	Registry of source component registrations.
sink_registry	ComponentRegistry[Sink]	Registry of sink component registrations.
processor_registry	ComponentRegistry[Processor]	Registry of processor component registrations.
sql_engine_factory	SqlEngineFactory	Shared provider for preflighted SQL connection references and the lifecycle of created SQL engines.
config_validator	ConfigValidator	Validator for the root pipeline configuration and its component sections.

etlrelay.default_components.create_default_component_catalog(*, sql_engine_factory=None)

Create the default component catalog and its shared dependencies.

The catalog registers CSV and Parquet readers and writers, local filesystem storage, file and SQL sources and sinks, and the select_columns, rename_columns, and filter_rows processors. It also assembles the dispatchers and root configuration validator.

A single set of registries is shared between the corresponding validators and factories, and the same SQL engine factory is supplied to both SQL component factories.

Parameters:

Name	Type	Description	Default
sql_engine_factory	SqlEngineFactory | None	Optional SQL engine factory to share with the catalog's SQL source and sink factories. If omitted, a new SqlEngineFactory is created.	None

Returns:

Type	Description
ComponentCatalog	A ComponentCatalog containing the default component registries,
ComponentCatalog	shared SQL engine factory, and root configuration validator.

Note

Creating the catalog assembles configuration components and their registrations; it does not invoke component build methods, resolve SQL environment variables, create database engines, open files, or execute a pipeline.

The tabular file-format registry is separate from key-value document readers such as YamlReader, which are not registered as tabular data formats.

etlrelay.config.registry.ComponentRegistration dataclass

Associate a component configuration validator with its factory.

A registration defines how a component's configuration is validated and how its runtime implementation is constructed.

Attributes:

Name	Type	Description
validator	ComponentConfigValidator	Validator used to check the component's configuration.
factory	ComponentFactory[T]	Callable that constructs the runtime component from its configuration mapping.

Factories are stored without being invoked. Their invocation is the responsibility of the component builder.

etlrelay.config.registry.ComponentRegistry

Provide read-only lookup of named component registrations.

The registry copies the supplied mapping during initialization, preventing subsequent changes to the original mapping from affecting its contents. Registration names must not be empty or contain only whitespace.

Attributes:

Name	Type	Description
_registration		Mapping of component names to their registrations.

__init__(registration)

Initialize the registry from a mapping of names to registrations.

Parameters:

Name	Type	Description	Default
registration	Mapping[str, ComponentRegistration[T]]	Mapping of component names to their registrations.	required

Raises:

Type	Description
ValueError	If any registration name is empty or contains only whitespace.

get(name)

Return the registration associated with a component name.

Parameters:

Name	Type	Description	Default
name	str	Name of the component registration to retrieve.	required

Returns:

Type	Description
ComponentRegistration[T]	The registration associated with name.

Raises:

Type	Description
KeyError	If name is not registered.

names()

Return the registered component names in insertion order.

Returns:

Type	Description
tuple[str, ...]	A tuple containing the names of all registered components.

etlrelay.config.component_factory.ComponentFactory

Bases: Protocol[ComponentT_co]

Construct a runtime component from validated configuration.

build(config)

Build and return a runtime component.

etlrelay.config.component_validator.ComponentConfigValidator

Bases: Protocol

Define the interface for validating component configuration.

Implementations validate configuration for a particular component type and report validation failures as error messages rather than returning a boolean status.

validate(config, path)

Validate component-specific configuration.

Parameters:

Name	Type	Description	Default
config	Any	Configuration value to validate.	required
path	str	Configuration path identifying the location being validated. Implementations use this path to contextualize error messages.	required

Returns:

Type	Description
list[str]	A list of validation error messages. An empty list indicates that
list[str]	validation succeeded.

Note

The specific validation rules depend on the implementation.

etlrelay.config.component_dispatcher.ComponentConfigDispatcher

Dispatch component configuration validation by registered type.

The dispatcher uses a component registry to resolve the validator associated with a configuration's type field. It rejects malformed configurations, missing or invalid type fields, and unregistered component types before delegating to the registered validator.

__init__(registry, component_kind)

Initialize the dispatcher with a registry and component category.

Parameters:

Name	Type	Description	Default
registry	ComponentRegistry[Any]	Registry containing the component registrations and their configuration validators.	required
component_kind	str	Human-readable component category used in validation error messages, such as "source" or "processor".	required

validate(config, path)

Validate a component configuration using its registered validator.

The method first checks that the configuration is a mapping with valid string keys and contains a non-empty string type field. It then resolves the component registration and delegates validation to its registered validator. Validation stops at the first dispatch-level failure.

Parameters:

Name	Type	Description	Default
config	object	Component configuration value to validate.	required
path	str	Configuration path used to identify the location of validation errors.	required

Returns:

Type	Description
list[str]	A list of validation error messages. The list contains errors for
list[str]	malformed mappings, invalid or missing type fields, unregistered
list[str]	component types, or errors reported by the registered validator.
list[str]	An empty list indicates that validation succeeded.

Note

This method validates configuration only. It does not invoke the registered component factory or construct the component.

etlrelay.config.pipeline_builder.PipelineBuilder

Build runtime pipelines from validated configuration mappings.

The builder resolves source, processor, and sink registrations before invoking component factories. SQL environment references are preflighted before component construction to prevent missing or empty environment variables from causing partial pipeline construction.

getenv provides a lightweight environment lookup for isolated builder use and tests. In the default application composition, the shared SqlEngineFactory should be supplied so successful preflight makes resolved connection URLs available to SQL component factories.

The builder does not execute the pipeline it constructs. When a shared SQL engine factory is supplied, the caller is responsible for closing the builder after pipeline execution to release the factory's resources.

__init__(source_registry, sink_registry, processor_registry, getenv=None, *, sql_engine_factory=None)

Initialize the builder with component registries and SQL preflight.

Parameters:

Name	Type	Description	Default
source_registry	ComponentRegistry[Source]	Registry of source component registrations.	required
sink_registry	ComponentRegistry[Sink]	Registry of sink component registrations.	required
processor_registry	ComponentRegistry[Processor]	Registry of processor component registrations.	required
getenv	EnvironmentLookup | None	Optional environment-variable lookup function. Defaults to os.environ.get when omitted.	None
sql_engine_factory	SqlEnginePreflighter | None	Optional shared SQL engine factory interface used to preflight connection references and release SQL resources. When omitted, the builder uses getenv to preflight the required environment variables without managing SQL engine resources itself.	None

build(validated_config)

Construct a pipeline from a validated configuration mapping.

The builder resolves the source, sink, and all processor registrations before performing SQL environment preflight. Only after these steps succeed does it invoke component factories and construct the pipeline. Processors are constructed in their configured order.

The returned pipeline is not executed. If component construction fails after SQL resources have been acquired during this attempt, the shared SQL engine factory is closed before the original exception is propagated.

Parameters:

Name	Type	Description	Default
validated_config	Mapping[str, object]	Configuration mapping that has already passed configuration validation.	required

Returns:

Type	Description
Pipeline	A constructed Pipeline containing the configured source,
Pipeline	processors, and sink.

Raises:

Type	Description
PipelineBuildError	If required configuration sections are malformed, processor configurations are not mappings, component types are invalid or unregistered, SQL connection references are malformed, or SQL environment preflight fails.
Exception	Exceptions raised by component factories or pipeline construction propagate to the caller. If construction fails after SQL preflight and resource acquisition, cleanup is attempted before the original exception is re-raised.

close()

Release SQL resources owned by the shared engine factory.

Call this method after the built pipeline has finished executing. It does not close the pipeline's source, processors, or sink directly. When no SQL engine factory was supplied, this method does nothing.

Raises:

Type	Description
PipelineBuildError	If the SQL engine factory raises SqlEngineFactoryError while releasing its resources.

etlrelay.config.pipeline_builder.PipelineBuildError

Bases: ValueError

Raised when configuration cannot be composed into a runtime pipeline.

This exception represents structural, component-resolution, and configuration-preflight failures encountered while building a pipeline.

etlrelay.config.sql_engine_factory.SqlEngineProvider

Bases: Protocol

Define the minimal interface required by SQL component factories.

Implementations provide SQLAlchemy engines associated with connection references that have already passed environment-variable preflight.

create(variable_name)

Return the engine associated with a preflighted variable name.

Parameters:

Name	Type	Description	Default
variable_name	str	Name of the environment variable whose resolved connection URL identifies the engine.	required

Returns:

Type	Description
Engine	The SQLAlchemy engine associated with the specified variable name.

Raises:

Type	Description
SqlEngineFactoryError	If the implementation cannot provide an engine for the requested variable name.

etlrelay.config.sql_engine_factory.SqlEngineFactory

Resolve SQL connection URLs and manage SQLAlchemy engine lifecycles.

Environment-variable references must pass preflight before engines can be created. Successful preflight stores resolved connection URLs without creating engines. Engines are created on demand and cached by environment variable name.

The factory owns the engines it creates and disposes of them when close() is called. Closing also discards resolved connection URLs. Environment lookup and engine creation can be injected for testing.

__init__(getenv=None, engine_creator=None)

Initialize the factory with optional environment and engine adapters.

Parameters:

Name	Type	Description	Default
getenv	EnvironmentLookup | None	Callable used to retrieve environment variables. Defaults to os.environ.get when omitted.	None
engine_creator	EngineCreator | None	Callable used to create SQLAlchemy engines from resolved connection URLs. Defaults to sqlalchemy.create_engine when omitted.	None

preflight(references)

Resolve and validate all referenced connection environment variables.

The mapping associates configuration paths with environment-variable names. Each unique variable is read once, and missing or empty values are aggregated into a single error. No engines are created during preflight.

A successful preflight replaces any previous resolved URL set. A failed preflight leaves resolved URLs unavailable for engine creation. Preflight cannot be repeated while the factory has active engines.

Parameters:

Name	Type	Description	Default
references	Mapping[str, str]	Mapping from configuration paths to environment-variable names that provide SQL connection URLs.	required

Raises:

Type	Description
SqlEngineFactoryError	If engines are already active, a reference has an invalid path or variable name, or one or more referenced environment variables are missing or empty.

Note

Exceptions raised by the environment lookup callable are treated as missing environment-variable values. Error messages do not include resolved URLs or environment lookup exception details.

create(variable_name)

Create or retrieve the engine for a successfully preflighted variable.

Engines are created on demand and cached by environment-variable name. Subsequent requests for the same name return the cached engine rather than creating another instance.

Parameters:

Name	Type	Description	Default
variable_name	str	Name of the environment variable whose connection URL was resolved during successful preflight.	required

Returns:

Type	Description
Engine	The cached SQLAlchemy engine for the variable, or a newly created
Engine	engine that is stored for subsequent requests.

Raises:

Type	Description
SqlEngineFactoryError	If successful preflight has not occurred, the variable was not included in preflight, or engine creation fails. Engine creation errors identify the exception type without exposing the underlying exception message or connection URL.

close()

Dispose of owned engines and discard resolved connection URLs.

Engine disposal is attempted in reverse creation order. The factory clears its engine cache and resolved URLs before disposal, and continues disposing remaining engines if one fails. Calling this method more than once is safe.

Raises:

Type	Description
SqlEngineFactoryError	If one or more engines cannot be disposed. The error does not include underlying disposal exception details.

etlrelay.config.sql_engine_factory.SqlEngineFactoryError

Bases: ValueError

Raised when SQL environment preflight or engine management fails.

This exception represents failures such as missing connection configuration, engine creation errors, and engine disposal errors. Error messages are designed to avoid exposing connection URLs or credentials.

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.

Sources expose batches through an iterable. Implementations determine how underlying resources are acquired and released.

The Pipeline consumes the iterator returned by read() and explicitly closes it on termination if it provides a callable close() method. Source implementations that retain resources during iteration should support deterministic cleanup when iteration completes, fails, or is explicitly closed.

read()

Return an iterable of batches obtained from the source.

Resource acquisition may be lazy and occur during iteration rather than when this method is called.

Direct consumers are responsible for closing active closeable iterators when they stop consumption early. Pipeline manages this cleanup when executing the source.

Returns:

Type	Description
Iterable[Batch]	An iterable producing Batch objects.

Raises:

Type	Description
Exception	Implementation-specific access, reading, or parsing failures propagate to the caller.

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 data batches.

Concrete sinks deliver batches to their destinations and manage resources they acquire for the write operation.

Persistence, transaction, overwrite, and empty-batch semantics are defined by each sink implementation rather than this protocol.

write(batches)

Consume batches and write them to the destination.

Implementations may consume batches lazily. They must release resources they own when writing completes or fails.

A sink is not required to consume every batch before returning. The Pipeline manages cleanup of active source iterators when execution terminates.

Parameters:

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

Raises:

Type	Description
Exception	Implementation-specific validation, serialization, storage, or database failures propagate to the caller.

Note

Atomicity, rollback, overwrite behavior, and handling of empty batches depend on the concrete sink implementation.

etlrelay.core.reader.TabularReader

Bases: Protocol

Define the interface for reading tabular data from a binary stream.

Implementations parse format-specific input into ETLRelay batches, independently of the storage system providing the stream.

read(stream)

Read serialized tabular data as an iterable of batches.

Parameters:

Name	Type	Description	Default
stream	BinaryIO	An open, readable binary stream containing serialized tabular data.	required

Returns:

Type	Description
Iterable[Batch]	An iterable of parsed batches. Parsing may occur lazily during
Iterable[Batch]	iteration.

Note

The caller owns the supplied stream and must keep it open while consuming the returned iterable. Implementations must not close the stream, including when parsing or iteration fails.

Stream requirements, empty-input behavior, batch boundaries, and parsing errors are documented by each implementation. Some formats require a seekable stream.

etlrelay.core.writer.TabularWriter

Bases: Protocol

Define the interface for writing tabular batches to a binary stream.

Implementations perform format-specific serialization independently of filesystem access. The supplied stream remains owned by the caller.

write(batches, stream)

Serialize batches into one output stream.

Parameters:

Name	Type	Description	Default
batches	Iterable[Batch]	An iterable of batches to serialize.	required
stream	BinaryIO	An open, writable binary stream receiving serialized data.	required

Note

The caller owns the output stream. Implementations must not close it, including when serialization or batch iteration fails.

The method consumes batches synchronously. It may leave partial output if serialization or iteration fails; this protocol does not provide atomic writes or rollback.

Empty-input behavior, schema compatibility, stream requirements, and serialization errors are documented by each implementation.

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.

Pipeline owns the consumption lifecycle of the source iterator for the duration of run(). When execution terminates, it attempts to close both the processing iterator and the underlying source iterator when they provide callable close() methods.

This cleanup occurs after successful execution, early sink termination, or failures during source iteration, processing, or sink execution.

Execution exceptions take precedence over iterator cleanup failures. Secondary cleanup failures are logged without their exception messages. If execution otherwise succeeds but cleanup fails, the first cleanup exception propagates.

Pipeline does not own or dispose the underlying storage backends, database engines, or component instances.

__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 processed lazily and passed to the sink without buffering the entire input. If the source produces no batches, the sink is not called.

Close active source and processing iterators when execution terminates, including when the sink stops consuming batches early.

Exceptions from the source, processors, or sink propagate unchanged. If iterator cleanup also fails, the original execution exception takes precedence. Cleanup failures are logged without exposing their messages.

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 or ambiguous 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 is missing or ambiguous 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 rejects path traversal and symlink-based escapes at validation time.

Validation does not require the target to exist or guarantee that filesystem state remains unchanged after validation.

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 resolves outside the permitted base directory.

validate(path)

Resolve a path and ensure it remains within the permitted base directory.

Parameters:

Name	Type	Description	Default
path	Path | str	Relative or absolute path to validate.	required

Returns:

Type	Description
Path	The resolved path, which does not need to exist yet.

Raises:

Type	Description
PathValidationError	If the resolved path is outside the permitted base directory.

Note

This method checks path containment, not existence or access permissions. A subsequent filesystem operation can still fail, and filesystem state can change after validation.

etlrelay.storage.filesystem.FileSystemError

Bases: Exception

Represent a missing-input error in the filesystem contract.

This exception is raised when an input resource does not exist. Other filesystem errors, including permission and I/O errors, should propagate as their original exceptions rather than being wrapped in FileSystemError.

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.

The returned stream is owned by the caller, which is responsible for closing it.

Parameters:

Name	Type	Description	Default
path	str	Path or resource identifier understood by the implementation.	required

Returns:

Type	Description
BinaryIO	A readable binary stream.

Raises:

Type	Description
FileSystemError	If the input resource does not exist.
OSError	For other filesystem errors, including permission or I/O errors.

open_write(path)

Open a resource for binary writing.

The returned stream is owned by the caller, which is responsible for closing it. The operation uses write-mode semantics rather than append semantics. For local files, an existing destination is truncated and a new destination is created if its parent directory exists. Parent directories are not created automatically.

Parameters:

Name	Type	Description	Default
path	str	Path or resource identifier understood by the implementation.	required

Returns:

Type	Description
BinaryIO	A writable binary stream.

Raises:

Type	Description
OSError	If the destination 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 in binary read mode.

The path is validated before opening. A missing input is reported as FileSystemError, including when the file disappears between path validation and the open operation. Path-validation errors and other filesystem errors are not converted into FileSystemError.

Parameters:

Name	Type	Description	Default
path	str	Path 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 it.

Raises:

Type	Description
PathValidationError	If the resolved path is outside the permitted base directory.
FileSystemError	If the input file does not exist.
OSError	For permission and other filesystem errors not representing a missing input.

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	TabularReader	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	TabularReader	The reader used to parse the file contents.	required

read()

Yield batches parsed from the configured input file.

File access and reading occur when the returned generator is consumed. The source opens the stream through the configured filesystem and passes it to the configured reader. The source manages the stream lifetime and closes it when iteration completes, reading raises an exception, or the generator is explicitly closed.

Consumers that stop iteration early should explicitly close the generator when deterministic cleanup is required.

Exceptions raised by the filesystem or reader propagate to the caller without being converted into format-independent errors.

Returns:

Type	Description
Iterable[Batch]	An iterable of batches parsed by the configured reader.

Note: FileSource owns the binary stream opened through FileSystem, but does not own the FileSystem or TabularReader instances.

When consumed by Pipeline, the source iterator is explicitly
closed when pipeline execution terminates.

When consumed directly, callers stopping iteration early must
explicitly close the generator for deterministic stream cleanup.

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 TabularWriter 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	TabularWriter	TabularWriter 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	TabularWriter	TabularWriter 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.

Note: FileSink owns and closes the stream returned by FileSystem.open_write(). It does not close or otherwise manage the supplied FileSystem or TabularWriter instances.

With LocalFileSystem, an existing destination is truncated when
opened for writing. If serialization fails, an empty or partially
written file may remain. Previous contents are not restored.

File writes are not atomic and are not rolled back.
Format-specific empty-input behavior is determined by the TabularWriter.

SQL source and sink

etlrelay.sources.sql.SqlSource

Read SQL query results into an Arrow-backed batch.

The source executes a SQL query using a caller-provided SQLAlchemy engine and converts the results into a single :class:Batch. All result rows are materialized in memory before the batch is constructed.

An empty result set produces no batches. The source does not own or dispose of the engine supplied by the caller.

__init__(engine, query, *, parameters=None)

Initialize the SQL source.

Parameters:

Name	Type	Description	Default
engine	Engine	SQLAlchemy engine used to establish database connections. The caller retains responsibility for its lifecycle.	required
query	str	SQL query to execute. Named bind parameters can be specified using SQLAlchemy's :name syntax.	required
parameters	Mapping[str, Any] | None	Optional mapping of named bind parameters to their values. Values are bound separately from the SQL statement.	None

read()

Execute the query and return its results as batches.

Returns:

Type	Description
Iterable[Batch]	An iterable containing one batch if the query returns rows,
Iterable[Batch]	or an empty iterable if the result set is empty.

Raises:

Type	Description
ValueError	If the query returns duplicate column names.
SQLAlchemyError	If query execution fails.

Note

The query result and connection are closed after row retrieval, including when retrieval fails. Arrow conversion occurs after the connection is released and may raise Arrow conversion errors. All rows are materialized in memory; this is not a streaming query. The supplied engine remains owned by the caller and is not disposed.

etlrelay.sinks.sql.SqlSink

Append Arrow-backed batches to an existing SQL table.

The sink uses SQLAlchemy Core to construct and execute insert statements. Each non-empty batch is written in its own transaction, allowing successful batches to remain committed if a subsequent batch fails.

The sink does not own or dispose of the supplied engine.

__init__(engine, table_name, *, schema=None)

Initialize the SQL sink.

Parameters:

Name	Type	Description	Default
engine	Engine	SQLAlchemy engine used to connect to the target database. The caller retains responsibility for its lifecycle.	required
table_name	str	Name of the existing database table to which batches will be appended.	required
schema	str | None	Optional database schema containing the target table.	None

write(batches)

Append batches to the target SQL table.

Each non-empty batch is inserted within its own transaction. Empty batches are skipped without opening a transaction. Batch columns are validated against the target table before insertion.

Previously committed batches remain committed if a later batch fails. A failure during a batch write rolls back that batch's transaction and propagates the exception to the caller.

Parameters:

Name	Type	Description	Default
batches	Iterable[Batch]	Iterable of Arrow-backed batches containing rows to append.	required

Raises:

Type	Description
ValueError	If a batch contains duplicate column names or columns that do not exist in the target table.
SQLAlchemyError	If table reflection or database execution fails, including missing tables and database constraint violations.

Note

SqlSink borrows the supplied SQLAlchemy Engine and does not dispose it.

Each nonempty batch is written in a separate transaction. A transaction commits only after its context manager exits successfully. Failed transactions are rolled back by SQLAlchemy, while previously committed batches remain committed.

Zero-row batches are skipped before reflection, column validation, or transaction creation.

Batch columns must be unique and exist in the destination table. Omitted destination columns are permitted when supported by database defaults, nullability, or other database rules.

SqlSink does not create tables, migrate schemas, upsert rows, or provide a transaction spanning multiple batches.

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.

Finalization is attempted after serialization or input iteration fails. The original exception takes precedence over a secondary finalization failure, which is logged without its message. If serialization succeeds, a finalization failure propagates. Partial output is not rolled back and may be unreadable if finalization fails.

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.

Additional public exceptions and registries

etlrelay.readers.yaml.YamlDocumentError

Bases: ValueError

Reject a YAML document that violates the root mapping contract.

Syntax and duplicate-key errors remain yaml.YAMLError instances. This ValueError subclass distinguishes expected document-shape failures from unrelated programming errors without changing existing ValueError handling by library callers.

etlrelay.core.reader.ReaderError

Bases: Exception

Represent a read or parsing failure normalized by a reader implementation.

etlrelay.core.writer.WriterError

Bases: Exception

Represent a serialization failure normalized by a writer implementation.

etlrelay.security.path.PathValidationError

Bases: Exception

Raised when a filesystem path fails validation.

etlrelay.file_formats.TabularFileFormatRegistry

Register and construct readers and writers for tabular formats.

Reading and writing capabilities are registered independently. A format may support reading, writing, both, or neither.

The registry copies its input mappings during initialization and preserves their insertion order. Factory callables are invoked only when the corresponding creation method is called.

__init__(readers, writers)

Initialize the registry with reader and writer factories.

Parameters:

Name	Type	Description	Default
readers	Mapping[str, Callable[[], ReaderT]]	Mapping of format names to reader factories.	required
writers	Mapping[str, Callable[[], WriterT]]	Mapping of format names to writer factories.	required

supports_reading(name)

Return whether a reader is registered for the given format.

Parameters:

Name	Type	Description	Default
name	str	Format name to check.	required

supports_writing(name)

Return whether a writer is registered for the given format.

Parameters:

Name	Type	Description	Default
name	str	Format name to check.	required

readable_formats()

Return readable format names in registration order.

writable_formats()

Return writable format names in registration order.

create_reader(name)

Construct a reader for a registered format.

Parameters:

Name	Type	Description	Default
name	str	Format name whose reader should be constructed.	required

Returns:

Type	Description
ReaderT	A new reader created by the registered factory.

Raises:

Type	Description
KeyError	If no reader is registered for the format.

create_writer(name)

Construct a writer for a registered format.

Parameters:

Name	Type	Description	Default
name	str	Format name whose writer should be constructed.	required

Returns:

Type	Description
WriterT	A new writer created by the registered factory.

Raises:

Type	Description
KeyError	If no writer is registered for the format.