reqwest stream!?

Cargo.toml

@@ -25,10 +25,15 @@ opt-level = 0
 
 [profile.release]
 opt-level = 3
-strip = true # strip debug symbols
+strip = true
 debug = false
-lto = "thin" # not sure if this should be 'fat'
-#lto = "fat"
+# __FAT_VS_THIN_LTO__?
+# tbd if this ought to be fat or thin, on one hand, 'thin' compiles MUCH faster
+# but on the other hand, I am an american...
+# there does not seem to be a noticeable difference in the performance tho
+# at the time of writing this (2024-12-20) the `fat` binary is 12.5mb and
+# the `thin` binary is 12.6mb (who's thin now!?).
+lto = "thin" # or "fat"
 
 # release fat profile
 [profile.bench]

crates/_ryo3-dev/src/lib.rs

@@ -24,8 +24,6 @@ pub mod sp;
 pub fn pymod_add(m: &Bound<'_, PyModule>) -> PyResult<()> {
     anystr::pymod_add(m)?;
     sp::pymod_add(m)?;
-
-    println!("adding req");
     ryo3_reqwest::pymod_add(m)?;
     Ok(())
 }

crates/ryo3-reqwest/Cargo.toml

@@ -14,11 +14,24 @@ pyo3 = { workspace = true, features = ["experimental-inspect", "experimental-asy
 pyo3-async-runtimes = { workspace = true, features = ["attributes", "tokio-runtime"] }
 tracing.workspace = true
 tokio.workspace = true
-reqwest = { version = "0.12.9", features = ["blocking", "brotli", "cookies", "gzip", "zstd", "charset", "http2", "macos-system-configuration", "rustls-tls-native-roots"], default-features = false}
+reqwest = { version = "0.12.9", features = [
+  "blocking",
+  "brotli",
+  "charset",
+  "cookies",
+  "gzip",
+  "http2",
+  "macos-system-configuration",
+  "rustls-tls-native-roots",
+  "stream",
+  "zstd"
+], default-features = false}
 bytes = { workspace = true }
 jiter.workspace = true
 ryo3-bytes.workspace = true
 ryo3-url.workspace = true
+futures-core = "0.3.31"
+futures-util = "0.3.31"
 
 [lints]
 workspace = true

crates/ryo3-reqwest/src/async_client.rs

@@ -1,12 +1,17 @@
-use pyo3::prelude::*;
-
 use crate::errors::map_reqwest_err;
 use crate::pyo3_bytes::Pyo3JsonBytes;
-use pyo3::exceptions::PyValueError;
+use bytes::Bytes;
+use futures_core::Stream;
+use futures_util::StreamExt;
+use pyo3::exceptions::{PyStopAsyncIteration, PyValueError};
+use pyo3::prelude::*;
 use pyo3::types::PyDict;
 use reqwest::StatusCode;
 use ryo3_bytes::Pyo3Bytes;
 use ryo3_url::PyUrl;
+use std::pin::Pin;
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 #[pyclass]
 #[pyo3(name = "AsyncClient")]
@@ -16,33 +21,35 @@ pub struct RyAsyncClient(reqwest::Client);
 #[pyo3(name = "AsyncResponse")]
 #[derive(Debug)]
 pub struct RyAsyncResponse {
-    // Store the response in an Option so we can take ownership later.
+    /// The actual response which will be consumed when read
+    res: Option<reqwest::Response>,
+
+    // ========================================================================
+    /// das status code
     status_code: StatusCode,
+    /// das headers
     headers: reqwest::header::HeaderMap,
-    // cookies: reqwest::cookie::CookieJar,
-    // version: Option<reqwest::Version>,
+    /// das url
     url: reqwest::Url,
-
-    // body: Option<Bytes>,
-    res: Option<reqwest::Response>,
+    /// das content length -- if it exists (tho it might not and/or be
+    /// different if the response is compressed)
+    content_length: Option<u64>,
 }
-// impl RyAsyncResponse {
-//     async fn read_body_async(&mut self) -> Result<(), PyErr> {
-//         if self.body.is_none() {
-//             let res = self
-//                 .res
-//                 .take()
-//                 .ok_or_else(|| PyValueError::new_err("Response already consumed"))?;
-//             let b = res
-//                 .bytes()
-//                 .await
-//                 .map_err(|e| PyValueError::new_err(format!("{e}")))?;
-//             self.body = Some(b);
-//         }
-//         Ok(())
-//     }
-// }
 
+impl From<reqwest::Response> for RyAsyncResponse {
+    fn from(res: reqwest::Response) -> Self {
+        Self {
+            status_code: res.status(),
+            headers: res.headers().clone(),
+            // cookies: res.cookies().clone(),
+            // version: res.version(),
+            url: res.url().clone(),
+            content_length: res.content_length(),
+            // body: None,
+            res: Some(res),
+        }
+    }
+}
 #[pymethods]
 impl RyAsyncResponse {
     #[getter]
@@ -67,12 +74,17 @@ impl RyAsyncResponse {
                 .to_str()
                 .map(String::from)
                 .map_err(|e| PyValueError::new_err(format!("{e}")))?;
-            // .to_str()?.to_string();
             pydict.set_item(k, v)?;
         }
         Ok(pydict)
     }
 
+    /// Return the content length of the response, if it is known or `None`.
+    #[getter]
+    fn content_length(&self) -> Option<u64> {
+        self.content_length
+    }
+
     fn bytes<'py>(&'py mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let response = self
             .res
@@ -110,6 +122,21 @@ impl RyAsyncResponse {
         })
     }
 
+    /// Return a response consuming async iterator over the response body
+    fn bytes_stream(&mut self) -> PyResult<RyAsyncResponseIter> {
+        let response = self
+            .res
+            .take()
+            .ok_or(PyValueError::new_err("Response already consumed"))?;
+
+        // HOLY SHIT THIS TOOK A LOT OF TRIAL AND ERROR
+        let stream = response.bytes_stream();
+        let stream = Box::pin(stream);
+        Ok(RyAsyncResponseIter {
+            stream: Arc::new(Mutex::new(stream)),
+        })
+    }
+
     fn __str__(&self) -> String {
         format!("Response: {}", self.status_code)
     }
@@ -119,28 +146,54 @@ impl RyAsyncResponse {
     }
 }
 
+// This whole response iterator was a difficult thing to figure out.
+//
+// NOTE: I (jesse) am pretty proud of this. I was struggling to get the
+//       async-iterator thingy to work bc rust + async is quite hard, but
+//       after lots and lots and lots of trial and error this works!
+//
+// clippy says this is too long and complicated to just sit in the struct def
+type AsyncResponseStreamInner =
+    Arc<Mutex<Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send>>>>;
+#[pyclass]
+pub struct RyAsyncResponseIter {
+    stream: AsyncResponseStreamInner,
+}
+
+#[pymethods]
+impl RyAsyncResponseIter {
+    fn __aiter__(this: PyRef<Self>) -> PyRef<Self> {
+        this
+    }
+
+    fn __anext__<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        let stream = self.stream.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            let mut guard = stream.lock().await;
+            match guard.as_mut().next().await {
+                Some(Ok(bytes)) => Ok(Some(Pyo3Bytes::from(bytes))),
+                Some(Err(e)) => Err(map_reqwest_err(e)),
+                // I totally forgot that this was a thing and that I couldn't just return None
+                None => Err(PyStopAsyncIteration::new_err("response-stream-fin")),
+            }
+        })
+    }
+}
+
 #[pymethods]
 impl RyAsyncClient {
     #[new]
     fn new() -> Self {
         Self(reqwest::Client::new())
     }
 
-    // self.request(Method::GET, url)
     fn get<'py>(&'py mut self, py: Python<'py>, url: &str) -> PyResult<Bound<'py, PyAny>> {
         let response_future = self.0.get(url).send();
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
-            let response = response_future
+            response_future
                 .await
-                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
-            let r = RyAsyncResponse {
-                status_code: response.status(),
-                headers: response.headers().clone(),
-                url: response.url().clone(),
-                // body: None,
-                res: Some(response),
-            };
-            Ok(r)
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
         })
     }
 
@@ -152,17 +205,60 @@ impl RyAsyncClient {
     ) -> PyResult<Bound<'py, PyAny>> {
         let response_future = self.0.post(url).body(body.to_vec()).send();
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
-            let response = response_future
+            response_future
                 .await
-                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
-            let r = RyAsyncResponse {
-                status_code: response.status(),
-                headers: response.headers().clone(),
-                url: response.url().clone(),
-                // body: None,
-                res: Some(response),
-            };
-            Ok(r)
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
+        })
+    }
+
+    fn put<'py>(
+        &'py mut self,
+        py: Python<'py>,
+        url: &str,
+        body: &[u8],
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let response_future = self.0.put(url).body(body.to_vec()).send();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            response_future
+                .await
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
+        })
+    }
+
+    fn patch<'py>(
+        &'py mut self,
+        py: Python<'py>,
+        url: &str,
+        body: &[u8],
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let response_future = self.0.patch(url).body(body.to_vec()).send();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            response_future
+                .await
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
+        })
+    }
+
+    fn delete<'py>(&'py mut self, py: Python<'py>, url: &str) -> PyResult<Bound<'py, PyAny>> {
+        let response_future = self.0.delete(url).send();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            response_future
+                .await
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
+        })
+    }
+
+    fn head<'py>(&'py mut self, py: Python<'py>, url: &str) -> PyResult<Bound<'py, PyAny>> {
+        let response_future = self.0.head(url).send();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            response_future
+                .await
+                .map(RyAsyncResponse::from)
+                .map_err(map_reqwest_err)
         })
     }
 }