Storage & Pipelines¶
from sqlspec.storage import (
AsyncStoragePipeline,
AsyncWritableBytesSink,
DirectStreamWriteProtocol,
PartitionStrategyConfig,
ResolvedStorageTarget,
StorageBridgeJob,
StorageCapabilities,
StorageDestination,
StorageFormat,
StorageRegistry,
StorageTelemetry,
SyncStoragePipeline,
WritableBytesSink,
create_storage_bridge_job,
encode_arrow_to_async_sink,
encode_arrow_to_sink,
get_storage_bridge_diagnostics,
get_storage_bridge_metrics,
reset_storage_bridge_metrics,
resolve_storage_path,
storage_registry,
supports_direct_stream_write,
)
Multi-backend object storage registry (local, fsspec, obstore) and Apache Arrow / Parquet / JSONL streaming storage pipelines. See Arrow & Storage for configuration and usage examples.
Storage Registry¶
StorageRegistry
¶
Global storage registry for named backend configurations.
Allows registering named storage backends that can be accessed from anywhere in your application. Backends are automatically selected based on URI scheme unless explicitly overridden.
register_alias(alias, uri, *, backend=None, base_path='', **kwargs)
¶
Register a named alias for a storage configuration.
get(uri_or_alias, *, backend=None, **kwargs)
¶
Get backend instance using URI-first routing with automatic backend selection.
is_alias_registered(alias)
¶
Check if a named alias is registered.
list_aliases()
¶
List all registered aliases.
clear_cache(uri_or_alias=None)
¶
Clear resolved backend cache.
clear()
¶
Clear all aliases and instances.
clear_instances()
¶
Clear only cached instances, keeping aliases.
clear_aliases()
¶
Clear only aliases, keeping cached instances.
resolve_storage_path(path, base_path='', protocol=FILE_PROTOCOL, strip_file_scheme=True)
¶
Resolve path relative to base_path with protocol-specific handling.
Storage Pipelines¶
SyncStoragePipeline
¶
Bases: _StoragePipelineBase
Pipeline coordinating storage registry operations and telemetry.
write_rows(rows, destination, *, format_hint=None, storage_options=None, overwrite=True)
¶
Write dictionary rows to storage using cached serializers.
write_arrow(table, destination, *, format_hint=None, storage_options=None, compression=None, write_options=None, overwrite=True)
¶
Write an Arrow table to storage using zero-copy buffers.
read_arrow(source, *, file_format, storage_options=None)
¶
Read an artifact from storage and decode it into an Arrow table.
stream_read(source, *, chunk_size=None, storage_options=None)
¶
Stream bytes from an artifact.
read_arrow_batches(source, *, file_format='parquet', batch_size=65536, arrow_schema=None, storage_options=None)
¶
Stream Arrow record batches from storage synchronously.
write_arrow_batches(destination, batches, *, file_format='parquet', compression=None, write_options=None, storage_options=None, overwrite=True)
¶
Write Arrow record batches to storage synchronously.
list_sources(source, *, file_format=None, storage_options=None)
¶
Expand a single URI, directory prefix, or glob pattern into sorted source URIs.
AsyncStoragePipeline
¶
Bases: _StoragePipelineBase
Async variant of the storage pipeline leveraging async-capable backends when available.
write_rows(rows, destination, *, format_hint=None, storage_options=None, overwrite=True)
¶
write_arrow(table, destination, *, format_hint=None, storage_options=None, compression=None, write_options=None, overwrite=True)
¶
read_arrow_async(source, *, file_format, storage_options=None)
¶
stream_read_async(source, *, chunk_size=None, storage_options=None)
¶
Stream bytes from an artifact asynchronously.
read_arrow_batches_async(source, *, file_format='parquet', batch_size=65536, arrow_schema=None, storage_options=None)
¶
Stream Arrow record batches from storage asynchronously.
write_arrow_batches_async(destination, batches, *, file_format='parquet', compression=None, write_options=None, storage_options=None, overwrite=True)
¶
Write Arrow record batches to storage asynchronously.
list_sources_async(source, *, file_format=None, storage_options=None)
¶
Expand a single URI, directory prefix, or glob pattern into sorted source URIs asynchronously.
Bridge Jobs, Capabilities & Telemetry¶
StorageBridgeJob
¶
Bases: TypedDict
Runtime-evaluated driver storage capabilities.
Attributes:
| Name | Type | Description |
|---|---|---|
arrow_export_enabled |
bool
|
Whether the driver supports exporting query results as Apache Arrow tables or batches. |
arrow_import_enabled |
bool
|
Whether the driver supports ingesting Apache Arrow tables or record batches. |
parquet_export_enabled |
bool
|
Whether direct or staged Parquet export is available. |
parquet_import_enabled |
bool
|
Whether direct or staged Parquet ingestion is available. |
partition_strategies |
list[str]
|
Supported partition fan-out strategy names (such as |
default_storage_profile |
NotRequired[str | None]
|
Named storage registry profile used when callers omit an explicit backend URI. |
Bases: TypedDict
Configuration for partition fan-out strategies.
Attributes:
| Name | Type | Description |
|---|---|---|
kind |
str
|
Partitioning strategy identifier ( |
partitions |
int
|
Target number of output partitions to produce. |
rows_per_chunk |
int
|
Maximum row count written per output file or batch chunk. |
manifest_path |
str
|
Optional destination URI or relative path for writing a partition manifest. |
ResolvedStorageTarget
¶
Bases: NamedTuple
A storage destination resolved to an address and its backend protocol.
Bases: TypedDict
Telemetry payload for storage bridge operations.
Attributes:
| Name | Type | Description |
|---|---|---|
destination |
str
|
Resolved target or source storage URI for the bridge transfer. |
bytes_processed |
int
|
Total uncompressed or encoded payload bytes read or written. |
rows_processed |
int
|
Total number of records transferred across the storage bridge. |
partitions_created |
int
|
Number of partitioned output objects written during export. |
duration_s |
float
|
Elapsed wall-clock transfer time in seconds. |
read_duration_s |
float
|
Elapsed time spent reading source storage or Arrow batches in seconds. |
encode_duration_s |
float
|
Elapsed time spent encoding batches into wire payloads in seconds. |
write_duration_s |
float
|
Elapsed time spent writing payloads to the database or storage sink in seconds. |
commit_duration_s |
float
|
Elapsed time spent committing the transaction or finalizing the sink in seconds. |
shards_processed |
int
|
Number of storage shards loaded during a multi-shard operation. |
shards_skipped |
int
|
Number of already-checkpointed storage shards skipped during resume. |
wal_bytes |
int
|
Estimated or observed Write-Ahead Log bytes generated during ingestion. |
format |
str
|
Serialized wire/file format (such as |
extra |
dict[str, object]
|
Adapter-specific diagnostic counters or native COPY options. |
backend |
str
|
Storage backend implementation name ( |
correlation_id |
str
|
Active request correlation identifier attached to the storage job. |
config |
str
|
Adapter configuration class name executing the storage transfer. |
bind_key |
str
|
Registered SQLSpec configuration bind key, when configured. |
create_storage_bridge_job(status, telemetry)
¶
Create a storage bridge job handle with a unique identifier.
get_storage_bridge_diagnostics()
¶
Return aggregated storage bridge + serializer cache metrics.
get_storage_bridge_metrics()
¶
Return aggregated storage bridge metrics.
reset_storage_bridge_metrics()
¶
Reset aggregated storage bridge metrics.
Direct Stream Writing & Arrow Sink Encoders¶
encode_arrow_to_sink(data, sink, *, format_hint='parquet', compression=None, write_options=None)
¶
Stream an Arrow table or record-batch source to a synchronous byte sink.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
ArrowTable | Any
|
Arrow Table, RecordBatch, RecordBatchReader, Arrow stream producer, or iterable of RecordBatch objects. |
required |
sink
|
Any
|
Writable byte sink accepting bytes chunks via write(data). |
required |
format_hint
|
StorageFormat
|
Target serialization format (parquet, arrow-ipc, or csv). |
'parquet'
|
compression
|
str | None
|
Optional codec name forwarded to Parquet or Arrow IPC writers. |
None
|
write_options
|
dict[str, Any] | None
|
Optional format-specific PyArrow writer options. |
None
|
Returns:
| Type | Description |
|---|---|
tuple[int, int]
|
Tuple of (bytes_written, rows_processed). |
Raises:
| Type | Description |
|---|---|
StorageCapabilityError
|
If format_hint is not a supported Arrow write format. |
encode_arrow_to_async_sink(data, sink, *, format_hint='parquet', compression=None, write_options=None, chunk_size=1048576)
¶
Stream an Arrow table or record-batch source to an async byte sink with bounded buffering.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
ArrowTable | Any
|
Arrow Table, RecordBatch, RecordBatchReader, Arrow stream producer, or sync/async iterable of RecordBatch objects. |
required |
sink
|
AsyncWritableBytesSink
|
Asynchronous writable byte sink receiving coalesced byte chunks. |
required |
format_hint
|
StorageFormat
|
Target serialization format (parquet, arrow-ipc, or csv). |
'parquet'
|
compression
|
str | None
|
Optional codec name forwarded to Parquet or Arrow IPC writers. |
None
|
write_options
|
dict[str, Any] | None
|
Optional format-specific PyArrow writer options. |
None
|
chunk_size
|
int
|
Coalesced staging chunk size in bytes before enqueuing to sink. |
1048576
|
Returns:
| Type | Description |
|---|---|
tuple[int, int]
|
Tuple of (bytes_written, rows_processed). |
Raises:
| Type | Description |
|---|---|
StorageCapabilityError
|
If format_hint is not a supported Arrow write format. |
ValueError
|
If chunk_size is not a positive integer. |
DirectStreamWriteProtocol
¶
Bases: Protocol
Optional capability protocol for backends supporting streaming Arrow writes.
supports_direct_arrow_write(file_format, *, overwrite=True)
¶
Return whether the backend can stream the requested Arrow format directly.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_format
|
StorageFormat
|
Target storage format (such as parquet, arrow-ipc, or csv). |
required |
overwrite
|
bool
|
Whether replacing an existing destination object is permitted. |
True
|
Returns:
| Type | Description |
|---|---|
bool
|
True when the backend supports direct streaming for the format and overwrite mode. |
open_write_stream_sync(path, *, overwrite=True, **kwargs)
¶
Open a synchronous streaming byte-sink session that aborts on unhandled exceptions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Destination object path or URI segment. |
required |
overwrite
|
bool
|
Whether replacing an existing destination object is permitted. |
True
|
**kwargs
|
Any
|
Backend-specific writer configuration options. |
{}
|
Returns:
| Type | Description |
|---|---|
AbstractContextManager[WritableBytesSink]
|
Context manager yielding a synchronous writable byte sink. |
open_write_stream_async(path, *, overwrite=True, **kwargs)
¶
Open an asynchronous streaming byte-sink session that aborts on errors or cancellation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
Destination object path or URI segment. |
required |
overwrite
|
bool
|
Whether replacing an existing destination object is permitted. |
True
|
**kwargs
|
Any
|
Backend-specific writer configuration options. |
{}
|
Returns:
| Type | Description |
|---|---|
AbstractAsyncContextManager[AsyncWritableBytesSink]
|
Async context manager yielding an asynchronous writable byte sink. |
WritableBytesSink
¶
AsyncWritableBytesSink
¶
supports_direct_stream_write(obj)
¶
Return whether a storage backend instance implements DirectStreamWriteProtocol.
Returns:
| Type | Description |
|---|---|
TypeGuard[DirectStreamWriteProtocol]
|
True if obj is an instance exposing direct synchronous and asynchronous write streams. |