Streaming uploads: Azure / GCS / Artifactory / SSH / FTP

rust/Cargo.toml

@@ -12,9 +12,9 @@ crate-type = ["cdylib"]
 [dependencies]
 pyo3 = { version = "0.29", features = ["extension-module"] }
 pyo3-async-runtimes = { version = "0.29", features = ["tokio-runtime"] }
-tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "fs"] }
 aws-config = { version = "1", features = ["behavior-version-latest"] }
-reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "http2"] }
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "http2", "stream"] }
 futures = "0.3"
 aws-sigv4 = "1"
 aws-credential-types = { version = "1", features = ["hardcoded-credentials"] }
@@ -34,6 +34,7 @@ scraper = { version = "0.27", default-features = false }
 url = "2"
 russh = "0.63"
 russh-sftp = "3"
+tokio-util = { version = "0.7", features = ["io"] }
 anyhow = "1"
 ssh2-config = { version = "0.8", default-features = false, features = ["nolog"] }
 

rust/asanypath_native.pyi

@@ -46,6 +46,8 @@ def s3_presign(
 # -- GCS --
 async def gcs_get_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def gcs_put_batch(*args: Any, **kwargs: Any) -> Any: ...
+async def gcs_start_resumable(*args: Any, **kwargs: Any) -> Any: ...
+async def gcs_upload_chunk(*args: Any, **kwargs: Any) -> Any: ...
 async def gcs_copy_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def gcs_delete_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def gcs_exists_batch(*args: Any, **kwargs: Any) -> Any: ...
@@ -66,6 +68,8 @@ def gcs_presign(
 # -- Azure --
 async def az_get_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def az_put_batch(*args: Any, **kwargs: Any) -> Any: ...
+async def az_put_block(*args: Any, **kwargs: Any) -> Any: ...
+async def az_put_block_list(*args: Any, **kwargs: Any) -> Any: ...
 async def az_copy_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def az_delete_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def az_exists_batch(*args: Any, **kwargs: Any) -> Any: ...
@@ -88,6 +92,7 @@ def az_presign(
 # -- Artifactory --
 async def art_get_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def art_put_batch(*args: Any, **kwargs: Any) -> Any: ...
+async def art_put_stream(*args: Any, **kwargs: Any) -> Any: ...
 async def art_copy_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def art_delete_batch(*args: Any, **kwargs: Any) -> Any: ...
 async def art_exists_batch(*args: Any, **kwargs: Any) -> Any: ...
@@ -114,6 +119,7 @@ async def range_read(path: str, start: int, end: int, **kwargs: Any) -> bytes: .
 async def ssh_read(*args: Any, **kwargs: Any) -> bytes: ...
 async def ssh_read_range(*args: Any, **kwargs: Any) -> bytes: ...
 async def ssh_write(*args: Any, **kwargs: Any) -> int: ...
+async def ssh_write_chunk(*args: Any, **kwargs: Any) -> int: ...
 async def ssh_exists(*args: Any, **kwargs: Any) -> bool: ...
 async def ssh_stat(*args: Any, **kwargs: Any) -> Any: ...
 async def ssh_lstat(*args: Any, **kwargs: Any) -> Any: ...

rust/src/artifactory.rs

@@ -19,7 +19,7 @@ use pyo3::prelude::*;
 use serde::Deserialize;
 use std::collections::HashMap;
 
-use crate::http::{bearer_headers, do_request_light, status_to_pyerr};
+use crate::http::{bearer_headers, do_request_light, do_request_stream_file, status_to_pyerr};
 
 // ---------------------------------------------------------------------------
 // JSON response types for Artifactory Storage API
@@ -465,6 +465,44 @@ pub fn art_put_batch<'py>(
     })
 }
 
+/// Stream-upload a single file by PUT (body framed from disk; bounded memory).
+/// Artifactory has no multipart API, so a large object is sent as one streamed PUT.
+#[pyfunction]
+#[pyo3(signature = (base_url, key, file_path, size, token, options_json=None, use_h2=None))]
+pub fn art_put_stream<'py>(
+    py: Python<'py>,
+    base_url: String,
+    key: String,
+    file_path: String,
+    size: u64,
+    token: String,
+    options_json: Option<String>,
+    use_h2: Option<bool>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let h2 = use_h2.unwrap_or(false);
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let options: crate::backend::UploadOptions =
+            sonic_rs::from_str(&options_json.unwrap_or_else(|| "{}".to_string()))
+                .map_err(|e| PyValueError::new_err(format!("invalid backend_options: {e}")))?;
+        crate::backend::require_empty_provider(&options)?;
+        let creds = art_creds!(base_url, token);
+        let url = crate::backend::append_query(&creds.content_url(&key), &options.query);
+        let mut headers = creds.auth_headers();
+        headers.push((
+            "content-type".to_string(),
+            "application/octet-stream".to_string(),
+        ));
+        crate::backend::merge_user_headers(&mut headers, &options)?;
+        let (status, body) = do_request_stream_file("PUT", &url, &headers, &file_path, size, h2)
+            .await
+            .map_err(PyRuntimeError::new_err)?;
+        if !status.is_success() {
+            return Err(status_to_pyerr(status, &creds.error_path(&key), &body));
+        }
+        Ok(())
+    })
+}
+
 /// Copy multiple files concurrently (server-side). Pairs are `(src, dst)`.
 #[pyfunction]
 pub fn art_copy_batch<'py>(

rust/src/azure.rs

@@ -943,6 +943,97 @@ pub fn az_put_batch<'py>(
     })
 }
 
+// ---------------------------------------------------------------------------
+// Block blob upload (orchestrated from Python): stage blocks, then commit list
+// ---------------------------------------------------------------------------
+
+/// Stage one block for a block blob. `block_id` is the raw (pre-URL-encode) id.
+#[pyfunction]
+#[pyo3(signature = (endpoint, container, blob, block_id, data, account_name, account_key=None, sas_token=None, use_h2=None))]
+pub fn az_put_block<'py>(
+    py: Python<'py>,
+    endpoint: String,
+    container: String,
+    blob: String,
+    block_id: String,
+    data: Vec<u8>,
+    account_name: String,
+    account_key: Option<String>,
+    sas_token: Option<String>,
+    use_h2: Option<bool>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let h2 = use_h2.unwrap_or(false);
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let creds = az_creds!(endpoint, container, account_name, account_key, sas_token);
+        let base = creds.blob_url(&blob);
+        let sep = if base.contains('?') { "&" } else { "?" };
+        let url = format!(
+            "{}{}comp=block&blockid={}",
+            base,
+            sep,
+            urlencoding::encode(&block_id)
+        );
+        let mut headers = vec![("content-length".to_string(), data.len().to_string())];
+        let params = [("blockid", block_id.as_str()), ("comp", "block")];
+        creds.sign("PUT", &url, &mut headers, Some(&params));
+        let (status, body) = do_request_light("PUT", &url, &headers, Some(Bytes::from(data)), h2)
+            .await
+            .map_err(PyRuntimeError::new_err)?;
+        if !status.is_success() {
+            return Err(status_to_pyerr(status, &creds.error_path(&blob), &body));
+        }
+        Ok(())
+    })
+}
+
+/// Commit a block blob from the staged block ids (in order).
+#[pyfunction]
+#[pyo3(signature = (endpoint, container, blob, block_ids, account_name, account_key=None, sas_token=None, options_json=None, use_h2=None))]
+pub fn az_put_block_list<'py>(
+    py: Python<'py>,
+    endpoint: String,
+    container: String,
+    blob: String,
+    block_ids: Vec<String>,
+    account_name: String,
+    account_key: Option<String>,
+    sas_token: Option<String>,
+    options_json: Option<String>,
+    use_h2: Option<bool>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let h2 = use_h2.unwrap_or(false);
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let options: crate::backend::UploadOptions =
+            sonic_rs::from_str(&options_json.unwrap_or_else(|| "{}".to_string()))
+                .map_err(|e| PyValueError::new_err(format!("invalid backend_options: {e}")))?;
+        crate::backend::require_empty_provider(&options)?;
+        let creds = az_creds!(endpoint, container, account_name, account_key, sas_token);
+        let base = creds.blob_url(&blob);
+        let sep = if base.contains('?') { "&" } else { "?" };
+        let url = format!("{}{}comp=blocklist", base, sep);
+        let mut xml = String::from("<?xml version=\"1.0\" encoding=\"utf-8\"?><BlockList>");
+        for id in &block_ids {
+            xml.push_str(&format!("<Latest>{}</Latest>", id));
+        }
+        xml.push_str("</BlockList>");
+        let body = Bytes::from(xml.into_bytes());
+        let mut headers = vec![
+            ("content-length".to_string(), body.len().to_string()),
+            ("content-type".to_string(), "application/xml".to_string()),
+        ];
+        crate::backend::merge_user_headers(&mut headers, &options)?;
+        let params = [("comp", "blocklist")];
+        creds.sign("PUT", &url, &mut headers, Some(&params));
+        let (status, resp) = do_request_light("PUT", &url, &headers, Some(body), h2)
+            .await
+            .map_err(PyRuntimeError::new_err)?;
+        if !status.is_success() {
+            return Err(status_to_pyerr(status, &creds.error_path(&blob), &resp));
+        }
+        Ok(())
+    })
+}
+
 #[pyfunction]
 pub fn az_copy_batch<'py>(
     py: Python<'py>,

rust/src/gcs.rs

@@ -24,7 +24,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Mutex, OnceLock};
 use std::time::{Duration, Instant};
 
-use crate::http::{bearer_headers, do_request_light, status_to_pyerr};
+use crate::http::{bearer_headers, do_request, do_request_light, status_to_pyerr};
 
 // ---------------------------------------------------------------------------
 // Typed JSON response structs (avoids Value allocations)
@@ -930,6 +930,84 @@ pub fn gcs_put_batch<'py>(
     })
 }
 
+// ---------------------------------------------------------------------------
+// Resumable upload (orchestrated from Python): init a session, then PUT ranges
+// ---------------------------------------------------------------------------
+
+/// Begin a resumable upload session; returns the session URI (Location header).
+#[pyfunction]
+#[pyo3(signature = (endpoint, bucket, object, total, access_token=None, use_h2=None))]
+pub fn gcs_start_resumable<'py>(
+    py: Python<'py>,
+    endpoint: String,
+    bucket: String,
+    object: String,
+    total: u64,
+    access_token: Option<String>,
+    use_h2: Option<bool>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let h2 = use_h2.unwrap_or(false);
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let creds = gcs_creds!(endpoint, bucket, access_token);
+        let url = append_params(
+            &creds.upload_base(),
+            &[("uploadType", "resumable"), ("name", &object)],
+        );
+        let mut headers = creds.auth_headers().await;
+        headers.push(("content-length".to_string(), "0".to_string()));
+        headers.push((
+            "x-upload-content-type".to_string(),
+            "application/octet-stream".to_string(),
+        ));
+        headers.push(("x-upload-content-length".to_string(), total.to_string()));
+        let (status, body, resp_headers) = do_request("POST", &url, &headers, None, h2)
+            .await
+            .map_err(PyRuntimeError::new_err)?;
+        if !status.is_success() {
+            return Err(status_to_pyerr(status, &creds.error_path(&object), &body));
+        }
+        resp_headers
+            .get("location")
+            .cloned()
+            .ok_or_else(|| PyRuntimeError::new_err("GCS resumable init: no Location header"))
+    })
+}
+
+/// Upload one chunk to a resumable session. GCS completes when offset+len == total.
+#[pyfunction]
+#[pyo3(signature = (session_uri, data, offset, total, use_h2=None))]
+pub fn gcs_upload_chunk<'py>(
+    py: Python<'py>,
+    session_uri: String,
+    data: Vec<u8>,
+    offset: u64,
+    total: u64,
+    use_h2: Option<bool>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let h2 = use_h2.unwrap_or(false);
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        let len = data.len() as u64;
+        let end = offset + len - 1;
+        let headers = vec![
+            ("content-length".to_string(), len.to_string()),
+            (
+                "content-range".to_string(),
+                format!("bytes {offset}-{end}/{total}"),
+            ),
+        ];
+        let (status, body, _h) =
+            do_request("PUT", &session_uri, &headers, Some(Bytes::from(data)), h2)
+                .await
+                .map_err(PyRuntimeError::new_err)?;
+        // 308 Resume Incomplete = intermediate chunk accepted; 2xx = final chunk done.
+        if status.as_u16() == 308 || status.is_success() {
+            Ok(())
+        } else {
+            Err(status_to_pyerr(status, "gs://resumable-chunk", &body))
+        }
+    })
+}
+
 /// Copy multiple objects concurrently (server-side). Pairs are `(src, dst)`.
 #[pyfunction]
 pub fn gcs_copy_batch<'py>(

rust/src/http.rs

@@ -158,6 +158,46 @@ pub async fn do_request_light(
     .await
 }
 
+/// Stream a request body from a file (bounded memory). An explicit
+/// `content_length` frames the body with Content-Length instead of chunked
+/// encoding; retries reopen the file so a consumed stream can be replayed.
+pub async fn do_request_stream_file(
+    method: &str,
+    url: &str,
+    headers: &[(String, String)],
+    file_path: &str,
+    content_length: u64,
+    use_h2: bool,
+) -> Result<(StatusCode, Bytes), String> {
+    let method_parsed = method.parse::<Method>().map_err(|e| e.to_string())?;
+    let client = get_client(use_h2);
+    let headers = headers.to_vec();
+
+    (async || {
+        let file = tokio::fs::File::open(file_path)
+            .await
+            .map_err(|e| e.to_string())?;
+        let stream = tokio_util::io::ReaderStream::new(file);
+        let mut req = client.request(method_parsed.clone(), url);
+        for (k, v) in &headers {
+            req = req.header(k.as_str(), v.as_str());
+        }
+        req = req
+            .header("content-length", content_length.to_string())
+            .body(reqwest::Body::wrap_stream(stream));
+        let resp = req.send().await.map_err(|e| e.to_string())?;
+        let status = resp.status();
+        if is_retryable_status(status) {
+            return Err(format!("HTTP {} (retryable)", status.as_u16()));
+        }
+        let resp_body = resp.bytes().await.map_err(|e| e.to_string())?;
+        Ok((status, resp_body))
+    })
+    .retry(retry_backoff())
+    .when(|e: &String| is_retryable(e))
+    .await
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------