chore(deps): bump object store crate (#16)

Signed-off-by: dark0dave <[email protected]>

Author: dark0dave

object-store-internal/Cargo.toml

@@ -10,7 +10,7 @@ async-trait = "0.1.57"
 bytes = "1.2.1"
 futures = "0.3"
 once_cell = "1.12.0"
-object_store = { version = "0.9", features = ["azure", "aws", "gcp"] }
+object_store = { version = "0.10.1", features = ["azure", "aws", "gcp"] }
 percent-encoding = "2"
 pyo3 = { version = "0.20", default-features = false, features = ["macros"] }
 pyo3-asyncio = { version = "0.20", features = ["tokio-runtime"] }

object-store-internal/src/file.rs

@@ -6,11 +6,10 @@ use crate::utils::{delete_dir, walk_tree};
 use crate::{ObjectStoreError, PyClientOptions};
 
 use object_store::path::Path;
-use object_store::{DynObjectStore, Error as InnerObjectStoreError, ListResult, MultipartId};
+use object_store::{DynObjectStore, Error as InnerObjectStoreError, ListResult, MultipartUpload};
 use pyo3::exceptions::{PyNotImplementedError, PyValueError};
 use pyo3::prelude::*;
 use pyo3::types::{IntoPyDict, PyBytes};
-use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio::runtime::Runtime;
 
 #[pyclass(subclass, weakref)]
@@ -442,11 +441,10 @@ impl ObjectInputFile {
 // TODO add buffer to store data ...
 #[pyclass(weakref)]
 pub struct ObjectOutputStream {
-    store: Arc<DynObjectStore>,
+    pub store: Arc<DynObjectStore>,
     rt: Arc<Runtime>,
-    path: Path,
-    writer: Box<dyn AsyncWrite + Send + Unpin>,
-    multipart_id: MultipartId,
+    pub path: Path,
+    writer: Box<dyn MultipartUpload>,
     pos: i64,
     #[pyo3(get)]
     closed: bool,
@@ -460,17 +458,18 @@ impl ObjectOutputStream {
         store: Arc<DynObjectStore>,
         path: Path,
     ) -> Result<Self, ObjectStoreError> {
-        let (multipart_id, writer) = store.put_multipart(&path).await.unwrap();
-        Ok(Self {
-            store,
-            rt,
-            path,
-            writer,
-            multipart_id,
-            pos: 0,
-            closed: false,
-            mode: "wb".into(),
-        })
+        match store.put_multipart(&path).await {
+            Ok(writer) => Ok(Self {
+                store,
+                rt,
+                path,
+                writer,
+                pos: 0,
+                closed: false,
+                mode: "wb".into(),
+            }),
+            Err(err) => Err(ObjectStoreError::ObjectStore(err)),
+        }
     }
 
     fn check_closed(&self) -> Result<(), ObjectStoreError> {
@@ -488,11 +487,11 @@ impl ObjectOutputStream {
 impl ObjectOutputStream {
     fn close(&mut self) -> PyResult<()> {
         self.closed = true;
-        match self.rt.block_on(self.writer.shutdown()) {
+        match self.rt.block_on(self.writer.complete()) {
             Ok(_) => Ok(()),
             Err(err) => {
                 self.rt
-                    .block_on(self.store.abort_multipart(&self.path, &self.multipart_id))
+                    .block_on(self.writer.abort())
                     .map_err(ObjectStoreError::from)?;
                 Err(ObjectStoreError::from(err).into())
             }
@@ -537,24 +536,25 @@ impl ObjectOutputStream {
 
     fn write(&mut self, data: &PyBytes) -> PyResult<i64> {
         self.check_closed()?;
-        let len = data.as_bytes().len() as i64;
-        match self.rt.block_on(self.writer.write_all(data.as_bytes())) {
+        let bytes = data.as_bytes().to_vec();
+        let len = bytes.len() as i64;
+        match self.rt.block_on(self.writer.put_part(bytes.into())) {
             Ok(_) => Ok(len),
             Err(err) => {
                 self.rt
-                    .block_on(self.store.abort_multipart(&self.path, &self.multipart_id))
+                    .block_on(self.writer.abort())
                     .map_err(ObjectStoreError::from)?;
                 Err(ObjectStoreError::from(err).into())
             }
         }
     }
 
     fn flush(&mut self) -> PyResult<()> {
-        match self.rt.block_on(self.writer.flush()) {
+        match self.rt.block_on(self.writer.complete()) {
             Ok(_) => Ok(()),
             Err(err) => {
                 self.rt
-                    .block_on(self.store.abort_multipart(&self.path, &self.multipart_id))
+                    .block_on(self.writer.abort())
                     .map_err(ObjectStoreError::from)?;
                 Err(ObjectStoreError::from(err).into())
             }