add try from uri with storage options

crates/deltalake-core/src/operations/mod.rs

@@ -7,6 +7,7 @@
 //! with a [data stream][datafusion::physical_plan::SendableRecordBatchStream],
 //! if the operation returns data as well.
 
+use std::collections::{HashMap};
 use self::create::CreateBuilder;
 use self::filesystem_check::FileSystemCheckBuilder;
 use self::vacuum::VacuumBuilder;
@@ -77,6 +78,18 @@ impl DeltaOps {
             Err(err) => Err(err),
         }
     }
+
+
+    /// try from uri with storage options
+    pub async fn try_from_uri_with_storage_options(uri: impl AsRef<str>, storage_options: HashMap<String, String>) -> DeltaResult<Self> {
+        let mut table = DeltaTableBuilder::from_uri(uri).with_storage_options(storage_options).build()?;
+        // We allow for uninitialized locations, since we may want to create the table
+        match table.load().await {
+            Ok(_) => Ok(table.into()),
+            Err(DeltaTableError::NotATable(_)) => Ok(table.into()),
+            Err(err) => Err(err),
+        }
+    }
 
     /// Create a new [`DeltaOps`] instance, backed by an un-initialized in memory table
     ///

python/src/lib.rs

@@ -1157,13 +1157,10 @@ fn write_to_deltalake(
 
     // let schema: StructType = (&schema.0).try_into().map_err(PythonError::from)?;
 
-
     let options = storage_options.clone().unwrap_or_default();
-    let table = DeltaTableBuilder::from_uri(table_uri).with_storage_options(options);
-
-    let table = table.build().map_err(PythonError::from)?;
+    let table = rt()?.block_on(DeltaOps::try_from_uri_with_storage_options(&table_uri, options)).map_err(PythonError::from)?;
 
-    let builder = DeltaOps(table)
+    let builder = table
         .write(batches)
         .with_save_mode(mode)
         .with_write_batch_size(max_rows_per_group as usize)