Python object urls

python/Cargo.toml

@@ -45,6 +45,16 @@ lightstream = { version = "0.7", path = "../rust", features = ["mmap", "uds"] }
 [features]
 default = []
 extension-module = ["pyo3/extension-module"]
+# Object storage URLs in place of file paths for `read` and `write`, with
+# the module-level `register_options` function. Enables `memory://` URLs,
+# and the backend features below enable their cloud schemes.
+object_storage = ["lightstream/object_storage"]
+# Amazon S3 and S3-compatible stores (`s3://`, `s3a://`)
+aws = ["object_storage", "lightstream/aws"]
+# Google Cloud Storage (`gs://`)
+gcp = ["object_storage", "lightstream/gcp"]
+# Azure Blob Storage (`az://`, `abfs://`, `abfss://`, `adl://`, `azure://`)
+azure = ["object_storage", "lightstream/azure"]
 
 [[example]]
 name = "tcp_server"

python/README.md

@@ -44,6 +44,31 @@ Arrow IPC readers use memory mapping and out-of-core routing where appropriate.
 
 Readers yield `minarrow.Table` batches. `read_all()` returns a `Table` when the result is contiguous or a `ChunkedTable` when multiple chunks must be preserved.
 
+### Object storage
+
+Builds with the `object_storage` feature accept object URLs wherever a file path is accepted. An object prefix with chunk objects beneath it reads in the same way as a chunk directory.
+
+```python
+ls.register_options("s3://telemetry", {"aws_region": "eu-west-2"})
+
+table = ls.read("s3://telemetry/trades/2026-10-14.parquet").read_all()
+
+with ls.write("s3://telemetry/trades/summary.arrow") as writer:
+    writer.write(table)
+```
+
+| Store                    | URL                               | Feature          |
+| ------------------------ | --------------------------------- | ---------------- |
+| In-process memory        | `memory:///path`                  | `object_storage` |
+| Amazon S3 and compatible | `s3://bucket/path`                | `aws`            |
+| Google Cloud Storage     | `gs://bucket/path`                | `gcp`            |
+| Azure Blob Storage       | `az://container/path`, `abfss://` | `azure`          |
+
+- Credentials come from each backend's standard environment variables.
+- `register_options` sets backend options, and the upload settings `part_size` and `max_concurrent_parts`, for every URL beneath a prefix.
+- Written objects become visible in their store when the writer closes.
+- Memory-mapped reads apply to local files only.
+
 ### Network transports
 
 The same interface streams raw Arrow IPC over TCP, Unix domain sockets, WebSocket, HTTP, QUIC, WebTransport and standard I/O.
@@ -225,6 +250,8 @@ maturin develop
 pytest tests/
 ```
 
+Optional features are enabled at build time, for example `maturin develop --features object_storage` or `maturin develop --features aws,gcp,azure`.
+
 ## Licence
 
 Copyright © 2025–2026 Peter Garfield Bower.

python/src/input.rs

@@ -11,7 +11,6 @@
 //! `Protocol` picks the wire framing, with both arms running over every
 //! transport, and each leaf variant wraps one lightstream end reader.
 
-use std::fs::File;
 use std::future::poll_fn;
 use std::io::{BufReader, Stdin};
 use std::path::PathBuf;
@@ -224,6 +223,9 @@ impl FileIO {
     /// Reads the next batch from the file, advancing the batch cursor for
     /// the indexed IPC readers and the chunk iterator for directories.
     fn read(&mut self) -> Result<Option<Table>, IoError> {
+        // Object URL reads wait on the embedded runtime from this thread.
+        #[cfg(feature = "object_storage")]
+        let _guard = runtime().enter();
         match self {
             FileIO::Ipc { reader, cursor } => {
                 if *cursor >= reader.num_batches() {
@@ -248,8 +250,8 @@ impl FileIO {
                     return Ok(None);
                 }
                 *done = true;
-                let file = File::open(&path).map_err(IoError::Io)?;
-                Ok(Some(load_parquet_table(file)?))
+                let reader = StorageReader::open(&path).map_err(IoError::Io)?;
+                Ok(Some(load_parquet_table(reader)?))
             }
             FileIO::ParquetChunked { reader } => reader.next().transpose(),
             FileIO::Csv { reader } => Ok(reader.next_batch()?),

python/src/lib.rs

@@ -19,6 +19,13 @@ mod runtime;
 mod tls;
 mod uri;
 
+#[cfg(feature = "object_storage")]
+use std::collections::HashMap;
+
+#[cfg(feature = "object_storage")]
+use lightstream::error::IoError;
+#[cfg(feature = "object_storage")]
+use lightstream::object_storage::register_options;
 use pyo3::prelude::*;
 
 pub use errors::{FormatError, LightstreamError, ProtocolError, TransportError, to_py_err};
@@ -27,16 +34,27 @@ pub use message::PyMessage;
 pub use output::PyDataStreamWriter;
 pub use runtime::{block_on_py, runtime};
 
-/// Opens a source for reading. The scheme picks the medium, `protocol`
-/// picks the wire protocol, and the format comes from `format` or the
-/// path extension. Returns a `Reader` that yields `minarrow.Table`
-/// batches. `accept=True` makes the reader the accepting peer on wire
-/// sources, holding the endpoint bound for the life of the process and
-/// blocking until a writer connects. `parallel` takes a stream count
-/// on TCP sources, where the reader always accepts that many
-/// connections. The quic, wt, wss, and https schemes carry TLS,
-/// taking `tls_cert` and `tls_key` PEM paths on the accepting peer
-/// and a `tls_ca` PEM path on the connecting peer.
+/// Opens a source for reading and returns a `Reader` that yields
+/// `minarrow.Table` batches.
+///
+/// The URI scheme selects the medium and `protocol` selects the wire
+/// protocol. The format comes from `format` or the path extension.
+///
+/// ## Wire sources
+/// - `accept=True` makes the reader the accepting peer. The endpoint stays
+///   bound for the life of the process, and the call blocks until a writer
+///   connects.
+/// - `parallel` sets the stream count on TCP sources, where the reader
+///   accepts that many connections.
+/// - The quic, wt, wss and https schemes use TLS. The accepting peer takes
+///   `tls_cert` and `tls_key` PEM paths, and the connecting peer takes a
+///   `tls_ca` PEM path.
+///
+/// ## Object storage
+/// Builds with the `object_storage` feature accept object URLs, such as
+/// `s3://bucket/key.parquet` or `memory:///key.parquet`, in place of file
+/// paths. An object prefix with chunk objects immediately beneath it reads
+/// as a chunk directory.
 #[pyfunction]
 #[pyo3(signature = (uri, *, format=None, protocol=None, parallel=0, accept=false, tls_cert=None, tls_key=None, tls_ca=None, mmap=None, out_of_core=false, base=None, delimiter=None, header=None, batch_size=None))]
 fn read(
@@ -77,14 +95,24 @@ fn read(
     Ok(PyDataStreamReader::new(resolved))
 }
 
-/// Opens a target for writing. The scheme picks the medium, `protocol`
-/// picks the wire protocol, and the format comes from `format` or the
-/// path extension. Returns a `Writer` accepting any Arrow-compatible
-/// object. `accept=True` makes the writer the accepting peer on wire
-/// targets, holding the endpoint bound for the life of the process and
-/// blocking until a reader connects. The quic, wt, wss, and https
-/// schemes carry TLS, taking `tls_cert` and `tls_key` PEM paths on
-/// the accepting peer and a `tls_ca` PEM path on the connecting peer.
+/// Opens a target for writing and returns a `Writer` that accepts any
+/// Arrow-compatible object.
+///
+/// The URI scheme selects the medium and `protocol` selects the wire
+/// protocol. The format comes from `format` or the path extension.
+///
+/// ## Wire targets
+/// - `accept=True` makes the writer the accepting peer. The endpoint stays
+///   bound for the life of the process, and the call blocks until a reader
+///   connects.
+/// - The quic, wt, wss and https schemes use TLS. The accepting peer takes
+///   `tls_cert` and `tls_key` PEM paths, and the connecting peer takes a
+///   `tls_ca` PEM path.
+///
+/// ## Object storage
+/// Builds with the `object_storage` feature accept object URLs in place of
+/// file paths. Each object becomes visible in its store when the writer
+/// closes.
 #[pyfunction]
 #[pyo3(signature = (uri, *, format=None, protocol=None, parallel=0, accept=false, tls_cert=None, tls_key=None, tls_ca=None, compression=None, delimiter=None, header=None))]
 fn write(
@@ -110,6 +138,39 @@ fn write(
     Ok(PyDataStreamWriter::new(resolved))
 }
 
+/// Registers object storage options for every object URL beneath `prefix`,
+/// for example `s3://telemetry` or `s3://telemetry/trades`.
+///
+/// Settings not registered here come from the backend's standard
+/// environment variables, such as `AWS_REGION`.
+///
+/// ## Options
+/// Keys and values are strings. The accepted keys are:
+/// - backend configuration keys, for example `aws_region`, `aws_endpoint`
+///   or `google_service_account`;
+/// - HTTP client keys, for example `timeout`;
+/// - the upload settings `part_size` and `max_concurrent_parts`.
+///
+/// ## Behaviour
+/// - Registering a prefix again replaces its earlier options.
+/// - Each URL uses the options of the longest registered prefix that
+///   contains it.
+///
+/// ## Errors
+/// - Raises `LightstreamError` when `prefix` is not an object URL or an
+///   upload setting is invalid.
+/// - Unknown backend keys raise `LightstreamError` when a URL beneath the
+///   prefix is first opened.
+#[cfg(feature = "object_storage")]
+#[pyfunction(name = "register_options")]
+fn register_object_options(prefix: &str, options: HashMap<String, String>) -> PyResult<()> {
+    let options: Vec<(&str, &str)> = options
+        .iter()
+        .map(|(key, value)| (key.as_str(), value.as_str()))
+        .collect();
+    register_options(prefix, &options).map_err(|e| to_py_err(IoError::Io(e)))
+}
+
 /// Streaming Arrow I/O for Python.
 #[pymodule]
 #[pyo3(name = "lightstream")]
@@ -122,6 +183,8 @@ fn lightstream_py(m: &Bound<'_, PyModule>) -> PyResult<()> {
 
     m.add_function(wrap_pyfunction!(read, m)?)?;
     m.add_function(wrap_pyfunction!(write, m)?)?;
+    #[cfg(feature = "object_storage")]
+    m.add_function(wrap_pyfunction!(register_object_options, m)?)?;
     m.add_class::<PyDataStreamReader>()?;
     m.add_class::<PyDataStreamWriter>()?;
     m.add_class::<PyMessage>()?;

python/src/output.rs

@@ -12,8 +12,7 @@
 //! over every transport, and each leaf variant wraps one lightstream end
 //! writer.
 
-use std::fs::File;
-use std::io;
+use std::io::{self, Write};
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
@@ -27,6 +26,7 @@ use lightstream::models::writers::csv::CsvWriter;
 use lightstream::models::writers::ipc::sync_table::SyncTableWriter;
 use lightstream::models::writers::json::JsonWriter;
 use lightstream::models::writers::lightstream::LightstreamWriter;
+use lightstream::models::writers::object_storage::StorageWriter;
 use lightstream::models::writers::parallel::tcp::TcpParallelTableWriter;
 use lightstream::models::writers::parquet::write_parquet_table;
 use lightstream::models::writers::http::HttpTableWriter;
@@ -86,7 +86,7 @@ pub enum FileIO {
     Ipc {
         path: PathBuf,
         compression: Option<Compression>,
-        writer: Option<SyncTableWriter<File>>,
+        writer: Option<SyncTableWriter<StorageWriter>>,
     },
     Parquet {
         path: PathBuf,
@@ -661,6 +661,9 @@ impl FileIO {
     /// from the first batch's schema, and the document formats buffer it
     /// for `finish`.
     fn write(&mut self, table: &Table) -> Result<(), IoError> {
+        // Object URL uploads wait on the embedded runtime from this thread.
+        #[cfg(feature = "object_storage")]
+        let _guard = runtime().enter();
         match self {
             FileIO::Ipc {
                 path,
@@ -670,7 +673,7 @@ impl FileIO {
                 if writer.is_none() {
                     let fields: Vec<Field> =
                         table.cols.iter().map(|col| (*col.field).clone()).collect();
-                    let file = File::create(&path).map_err(IoError::Io)?;
+                    let file = StorageWriter::create(&path).map_err(IoError::Io)?;
                     *writer = Some(SyncTableWriter::new(
                         file,
                         fields,
@@ -709,6 +712,9 @@ impl FileIO {
     /// document formats write their buffered batches as one file. A target
     /// that received no batches writes nothing.
     fn finish(&mut self) -> Result<(), IoError> {
+        // Object URL uploads wait on the embedded runtime from this thread.
+        #[cfg(feature = "object_storage")]
+        let _guard = runtime().enter();
         match self {
             FileIO::Ipc { writer, .. } => match writer.as_mut() {
                 Some(writer) => Ok(writer.finish()?),
@@ -724,8 +730,10 @@ impl FileIO {
                 }
                 let tables: Vec<Table> = batches.drain(..).map(|t| (*t).clone()).collect();
                 let table = tables.consolidate();
-                let file = File::create(&path).map_err(IoError::Io)?;
-                write_parquet_table(&table, file, *compression)
+                let mut file = StorageWriter::create(&path).map_err(IoError::Io)?;
+                write_parquet_table(&table, &mut file, *compression)?;
+                // Flushing completes the object when `path` is an object URL.
+                Ok(file.flush()?)
             }
             FileIO::CsvStdio { .. } | FileIO::JsonStdio { .. } => Ok(()),
             FileIO::Csv {
@@ -738,7 +746,7 @@ impl FileIO {
                 }
                 let name = Some(batches[0].name.clone());
                 let super_table = SuperTable::from_batches(batches.drain(..).collect(), name);
-                let file = File::create(&path).map_err(IoError::Io)?;
+                let file = StorageWriter::create(&path).map_err(IoError::Io)?;
                 let mut writer = CsvWriter::with_options(file, options.clone());
                 writer.write_supertable(&super_table)?;
                 Ok(writer.flush()?)
@@ -753,7 +761,7 @@ impl FileIO {
                 }
                 let name = Some(batches[0].name.clone());
                 let super_table = SuperTable::from_batches(batches.drain(..).collect(), name);
-                let file = File::create(&path).map_err(IoError::Io)?;
+                let file = StorageWriter::create(&path).map_err(IoError::Io)?;
                 let mut writer = JsonWriter::new(file, options.clone());
                 writer.write_supertable(&super_table)?;
                 Ok(writer.flush()?)
@@ -912,7 +920,8 @@ impl PyDataStreamWriter {
     }
 
     /// File targets write through on each batch, so flush delegates to the
-    /// operating system.
+    /// operating system. Objects at object URLs are completed on `close`,
+    /// which makes them visible in their store.
     fn flush(&self) -> PyResult<()> {
         Ok(())
     }

python/src/runtime.rs

@@ -7,8 +7,10 @@
 //! # Runtime
 //!
 //! The embedded Tokio runtime that drives the async readers and writers.
-//! File-format paths use the synchronous readers and writers and never
-//! touch it.
+//! File-format reads and writes use the synchronous readers and writers.
+//! With the `object_storage` feature, they enter the runtime on the
+//! calling Python thread, which is outside the runtime's executor, and
+//! object URL requests wait on the runtime through `Handle::block_on`.
 
 use std::sync::OnceLock;