Infer an http store for bare https urls (#575)

Follow up for #552.

Also closes #492. Closes
#551.

Author: kylebarron

python/core/python/geoarrow/rust/core/_rust.pyi

@@ -1466,8 +1466,8 @@ def read_flatgeobuf(
 ) -> GeoTable: ...
 async def read_flatgeobuf_async(
     path: str,
-    fs: ObjectStore,
     *,
+    fs: Optional[ObjectStore] = None,
     batch_size: int = 65536,
     bbox: Tuple[float, float, float, float] | None = None,
 ) -> GeoTable: ...
@@ -1479,7 +1479,12 @@ def read_geojson_lines(
 ) -> GeoTable: ...
 def read_ipc(file: Union[str, Path, BinaryIO]) -> GeoTable: ...
 def read_ipc_stream(file: Union[str, Path, BinaryIO]) -> GeoTable: ...
-def read_parquet(path: str, *, batch_size: int = 65536) -> GeoTable: ...
+def read_parquet(
+    path: str, *, fs: Optional[ObjectStore] = None, batch_size: int = 65536
+) -> GeoTable: ...
+async def read_parquet_async(
+    path: str, *, fs: Optional[ObjectStore] = None, batch_size: int = 65536
+) -> GeoTable: ...
 def read_postgis(connection_url: str, sql: str) -> Optional[GeoTable]: ...
 async def read_postgis_async(connection_url: str, sql: str) -> Optional[GeoTable]: ...
 def read_pyogrio(

python/core/src/error.rs

@@ -4,13 +4,19 @@ use pyo3::prelude::*;
 pub enum PyGeoArrowError {
     GeoArrowError(geoarrow::error::GeoArrowError),
     PyErr(PyErr),
+    ObjectStoreError(object_store::Error),
+    ObjectStorePathError(object_store::path::Error),
+    UrlParseError(url::ParseError),
 }
 
 impl From<PyGeoArrowError> for PyErr {
     fn from(error: PyGeoArrowError) -> Self {
         match error {
             PyGeoArrowError::GeoArrowError(err) => PyException::new_err(err.to_string()),
             PyGeoArrowError::PyErr(err) => err,
+            PyGeoArrowError::ObjectStoreError(err) => PyException::new_err(err.to_string()),
+            PyGeoArrowError::ObjectStorePathError(err) => PyException::new_err(err.to_string()),
+            PyGeoArrowError::UrlParseError(err) => PyException::new_err(err.to_string()),
         }
     }
 }
@@ -21,6 +27,24 @@ impl From<geoarrow::error::GeoArrowError> for PyGeoArrowError {
     }
 }
 
+impl From<object_store::Error> for PyGeoArrowError {
+    fn from(other: object_store::Error) -> Self {
+        Self::ObjectStoreError(other)
+    }
+}
+
+impl From<object_store::path::Error> for PyGeoArrowError {
+    fn from(other: object_store::path::Error) -> Self {
+        Self::ObjectStorePathError(other)
+    }
+}
+
+impl From<url::ParseError> for PyGeoArrowError {
+    fn from(other: url::ParseError) -> Self {
+        Self::UrlParseError(other)
+    }
+}
+
 impl From<PyTypeError> for PyGeoArrowError {
     fn from(other: PyTypeError) -> Self {
         Self::PyErr((&other).into())

python/core/src/io/csv.rs

@@ -1,5 +1,5 @@
 use crate::error::PyGeoArrowResult;
-use crate::io::file::{BinaryFileReader, BinaryFileWriter};
+use crate::io::input::sync::{BinaryFileReader, BinaryFileWriter};
 use crate::table::GeoTable;
 use geoarrow::io::csv::read_csv as _read_csv;
 use geoarrow::io::csv::write_csv as _write_csv;

python/core/src/io/flatgeobuf.rs

@@ -1,29 +1,56 @@
 use crate::error::{PyGeoArrowError, PyGeoArrowResult};
-use crate::io::file::{BinaryFileReader, BinaryFileWriter};
+use crate::io::input::sync::BinaryFileWriter;
+use crate::io::input::{construct_reader, FileReader};
 use crate::io::object_store::PyObjectStore;
 use crate::table::GeoTable;
 use flatgeobuf::FgbWriterOptions;
 use geoarrow::io::flatgeobuf::read_flatgeobuf_async as _read_flatgeobuf_async;
 use geoarrow::io::flatgeobuf::write_flatgeobuf_with_options as _write_flatgeobuf;
 use geoarrow::io::flatgeobuf::{read_flatgeobuf as _read_flatgeobuf, FlatGeobufReaderOptions};
+use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 
-/// Read a FlatGeobuf file from a path on disk into a GeoTable.
+/// Read a FlatGeobuf file from a path on disk or a remote location into a GeoTable.
 ///
 /// Example:
 ///
-/// Reading a remote file on an S3 bucket.
+/// Reading from a local path:
 ///
 /// ```py
-/// from geoarrow.rust.core import ObjectStore, read_flatgeobuf_async
+/// from geoarrow.rust.core import read_flatgeobuf
+/// table = read_flatgeobuf("path/to/file.fgb")
+/// ```
+///
+/// Reading from a Python file object:
+///
+/// ```py
+/// from geoarrow.rust.core import read_flatgeobuf
+///
+/// with open("path/to/file.fgb", "rb") as file:
+///     table = read_flatgeobuf(file)
+/// ```
+///
+/// Reading from an HTTP(S) url:
+///
+/// ```py
+/// from geoarrow.rust.core import read_flatgeobuf
+///
+/// url = "http://flatgeobuf.org/test/data/UScounties.fgb"
+/// table = read_flatgeobuf(url)
+/// ```
+///
+/// Reading from a remote file on an S3 bucket.
+///
+/// ```py
+/// from geoarrow.rust.core import ObjectStore, read_flatgeobuf
 ///
 /// options = {
 ///     "aws_access_key_id": "...",
 ///     "aws_secret_access_key": "...",
 ///     "aws_region": "..."
 /// }
 /// fs = ObjectStore('s3://bucket', options=options)
-/// table = read_flatgeobuf("path/in/bucket.fgb", fs)
+/// table = read_flatgeobuf("path/in/bucket.fgb", fs=fs)
 /// ```
 ///
 /// Args:
@@ -47,80 +74,102 @@ pub fn read_flatgeobuf(
     batch_size: usize,
     bbox: Option<(f64, f64, f64, f64)>,
 ) -> PyGeoArrowResult<GeoTable> {
-    if let Some(fs) = fs {
-        fs.rt.block_on(async move {
+    let reader = construct_reader(py, file, fs)?;
+    match reader {
+        FileReader::Async(async_reader) => async_reader.runtime.block_on(async move {
             let options = FlatGeobufReaderOptions {
                 batch_size: Some(batch_size),
                 bbox,
                 ..Default::default()
             };
-            let path = file.extract::<String>(py)?;
-            let table = _read_flatgeobuf_async(fs.inner, path.into(), options)
+            let table = _read_flatgeobuf_async(async_reader.store, async_reader.path, options)
                 .await
                 .map_err(PyGeoArrowError::GeoArrowError)?;
 
             Ok(GeoTable(table))
-        })
-    } else {
-        let mut reader = file.extract::<BinaryFileReader>(py)?;
-        let options = FlatGeobufReaderOptions {
-            batch_size: Some(batch_size),
-            bbox,
-            ..Default::default()
-        };
-        let table = _read_flatgeobuf(&mut reader, options)?;
-        Ok(GeoTable(table))
+        }),
+        FileReader::Sync(mut sync_reader) => {
+            let options = FlatGeobufReaderOptions {
+                batch_size: Some(batch_size),
+                bbox,
+                ..Default::default()
+            };
+            let table = _read_flatgeobuf(&mut sync_reader, options)?;
+            Ok(GeoTable(table))
+        }
     }
 }
 
 /// Read a FlatGeobuf file from a url into a GeoTable.
 ///
 /// Example:
 ///
-///     ```py
-///     from geoarrow.rust.core import ObjectStore, read_flatgeobuf_async
+/// Reading from an HTTP(S) url:
+///
+/// ```py
+/// from geoarrow.rust.core import read_flatgeobuf_async
+///
+/// url = "http://flatgeobuf.org/test/data/UScounties.fgb"
+/// table = await read_flatgeobuf_async(url)
+/// ```
 ///
-///     options = {
-///         "aws_access_key_id": "...",
-///         "aws_secret_access_key": "...",
-///     }
-///     fs = ObjectStore('s3://bucket', options=options)
-///     table = await read_flatgeobuf_async("path/in/bucket.fgb", fs)
-///     ```
+/// Reading from an S3 bucket:
+///
+/// ```py
+/// from geoarrow.rust.core import ObjectStore, read_flatgeobuf_async
+///
+/// options = {
+///     "aws_access_key_id": "...",
+///     "aws_secret_access_key": "...",
+///     "aws_region": "..."
+/// }
+/// fs = ObjectStore('s3://bucket', options=options)
+/// table = await read_flatgeobuf_async("path/in/bucket.fgb", fs=fs)
+/// ```
 ///
 /// Args:
-///     url: the url to a remote FlatGeobuf file
-///     fs: an ObjectStore instance for this url.
+///     path: the url or relative path to a remote FlatGeobuf file. If an argument is passed for
+///         `fs`, this should be a path fragment relative to the root passed to the `ObjectStore`
+///         constructor.
 ///
 /// Other args:
+///     fs: an ObjectStore instance for this url. This is required for non-HTTP urls.
 ///     batch_size: the number of rows to include in each internal batch of the table.
 ///     bbox: A spatial filter for reading rows, of the format (minx, miny, maxx, maxy). If set to
 ///       `None`, no spatial filtering will be performed.
 ///
 /// Returns:
 ///     Table from FlatGeobuf file.
 #[pyfunction]
-#[pyo3(signature = (path, fs, *, batch_size=65536, bbox=None))]
+#[pyo3(signature = (path, *, fs=None, batch_size=65536, bbox=None))]
 pub fn read_flatgeobuf_async(
     py: Python,
-    path: String,
-    fs: PyObjectStore,
+    path: PyObject,
+    fs: Option<PyObjectStore>,
     batch_size: usize,
     bbox: Option<(f64, f64, f64, f64)>,
 ) -> PyGeoArrowResult<PyObject> {
-    let fut = pyo3_asyncio::tokio::future_into_py(py, async move {
-        let options = FlatGeobufReaderOptions {
-            batch_size: Some(batch_size),
-            bbox,
-            ..Default::default()
-        };
-        let table = _read_flatgeobuf_async(fs.inner, path.into(), options)
-            .await
-            .map_err(PyGeoArrowError::GeoArrowError)?;
+    let reader = construct_reader(py, path, fs)?;
+    match reader {
+        FileReader::Async(async_reader) => {
+            let fut = pyo3_asyncio::tokio::future_into_py(py, async move {
+                let options = FlatGeobufReaderOptions {
+                    batch_size: Some(batch_size),
+                    bbox,
+                    ..Default::default()
+                };
+                let table = _read_flatgeobuf_async(async_reader.store, async_reader.path, options)
+                    .await
+                    .map_err(PyGeoArrowError::GeoArrowError)?;
 
-        Ok(GeoTable(table))
-    })?;
-    Ok(fut.into())
+                Ok(GeoTable(table))
+            })?;
+            Ok(fut.into())
+        }
+        FileReader::Sync(_) => {
+            Err(PyValueError::new_err("Local file paths not supported in async reader.").into())
+        }
+    }
 }
 
 /// Write a GeoTable to a FlatGeobuf file on disk.

python/core/src/io/geojson.rs

@@ -1,5 +1,5 @@
 use crate::error::PyGeoArrowResult;
-use crate::io::file::{BinaryFileReader, BinaryFileWriter};
+use crate::io::input::sync::{BinaryFileReader, BinaryFileWriter};
 use crate::table::GeoTable;
 use geoarrow::io::geojson::read_geojson as _read_geojson;
 use geoarrow::io::geojson::write_geojson as _write_geojson;

python/core/src/io/geojson_lines.rs

@@ -1,5 +1,5 @@
 use crate::error::PyGeoArrowResult;
-use crate::io::file::{BinaryFileReader, BinaryFileWriter};
+use crate::io::input::sync::{BinaryFileReader, BinaryFileWriter};
 use crate::table::GeoTable;
 use geoarrow::io::geojson_lines::read_geojson_lines as _read_geojson_lines;
 use geoarrow::io::geojson_lines::write_geojson_lines as _write_geojson_lines;

python/core/src/io/input/mod.rs

@@ -0,0 +1,77 @@
+pub mod sync;
+
+use std::sync::Arc;
+
+use crate::error::PyGeoArrowResult;
+use crate::io::object_store::PyObjectStore;
+use object_store::http::HttpBuilder;
+use object_store::path::Path;
+use object_store::{ClientOptions, ObjectStore};
+use pyo3::exceptions::PyValueError;
+use sync::BinaryFileReader;
+
+use pyo3::prelude::*;
+use tokio::runtime::Runtime;
+use url::Url;
+
+pub struct AsyncFileReader {
+    pub store: Arc<dyn ObjectStore>,
+    pub path: Path,
+    pub runtime: Arc<Runtime>,
+}
+
+pub enum FileReader {
+    Sync(BinaryFileReader),
+    Async(AsyncFileReader),
+}
+
+/// Construct a reader for the user that can be either synchronous or asynchronous
+///
+/// If the user has not passed in an object store instance but the `file` points to a http address,
+/// an HTTPStore will be created for it.
+pub fn construct_reader(
+    py: Python,
+    file: PyObject,
+    fs: Option<PyObjectStore>,
+) -> PyGeoArrowResult<FileReader> {
+    // If the user passed an object store instance, use that
+    if let Some(fs) = fs {
+        let path = file.extract::<String>(py)?;
+        let async_reader = AsyncFileReader {
+            store: fs.inner,
+            runtime: fs.rt,
+            path: path.into(),
+        };
+        Ok(FileReader::Async(async_reader))
+    } else {
+        // If the user's path is a "known" URL (i.e. http(s)) then construct an object store
+        // instance for them.
+        if let Ok(path_or_url) = file.extract::<String>(py) {
+            if path_or_url.starts_with("http") {
+                let url = Url::parse(&path_or_url)?;
+                // Expecting that the url input is something like
+                let store_input = format!("{}://{}", url.scheme(), url.domain().unwrap());
+
+                let options = ClientOptions::new().with_allow_http(true);
+                let store = HttpBuilder::new()
+                    .with_url(store_input)
+                    .with_client_options(options)
+                    .build()?;
+                let path = url.path().trim_start_matches('/');
+
+                let runtime = Arc::new(
+                    tokio::runtime::Runtime::new()
+                        .map_err(|err| PyValueError::new_err(err.to_string()))?,
+                );
+                let async_reader = AsyncFileReader {
+                    store: Arc::new(store),
+                    runtime,
+                    path: path.into(),
+                };
+                return Ok(FileReader::Async(async_reader));
+            }
+        }
+
+        Ok(FileReader::Sync(file.extract(py)?))
+    }
+}