API Reference

This reference documents ETLRelay's Python API using public classes, interfaces, and their docstrings.

Configuration APIs introduced in ETLRelay 0.6.0 are grouped separately from the direct runtime component APIs. The command-line interface introduced in ETLRelay 0.7.0 is documented in the CLI Reference.

The public API may evolve before the 1.0.0 release. Consult the configuration and architecture documentation for supported YAML properties, component combinations, and implementation boundaries.

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
ValueError	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

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

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.

The storage implementation may create or truncate the destination before serialization begins. If writing fails, the stream is closed but the output is not rolled back; a partial or empty file may remain. The original writer exception propagates to the caller.

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.

Note

The caller retains ownership of the stream and is responsible for managing its lifecycle. Implementations may propagate format-specific parsing and I/O exceptions; the protocol does not normalize them into ReaderError.

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)

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	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()

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.

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.

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.

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.

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.