Merge pull request #96 from influxdata/crepererum/http

feat: HTTP interface for Python guests

Author: crepererum

Cargo.lock

@@ -24,9 +24,10 @@ datafusion-udf-wasm-bundle = { path = "guests/bundle", version = "0.1.0" }
 datafusion-udf-wasm-guest = { path = "guests/rust", version = "0.1.0" }
 datafusion-udf-wasm-python = { path = "guests/python", version = "0.1.0" }
 tokio = { version = "1.48.0", default-features = false }
-pyo3 = { version = "0.27.0", default-features = false }
+pyo3 = { version = "0.27.0", default-features = false, features = ["macros"] }
 tar = { version = "0.4.44", default-features = false }
 tempfile = { version = "3.23.0", default-features = false }
+wasip2 = { version = "1" }
 wasmtime = { version = "37.0.2", default-features = false, features = ["async", "cranelift"] }
 wasmtime-wasi = { version = "37.0.0", default-features = false }
 wasmtime-wasi-http = { version = "37.0.0", default-features = false, features = ["default-send-request"] }

Cargo.toml

@@ -18,6 +18,7 @@ datafusion-common.workspace = true
 datafusion-expr.workspace = true
 datafusion-udf-wasm-guest.workspace = true
 pyo3.workspace = true
+wasip2.workspace = true
 
 [build-dependencies]
 tar.workspace = true

guests/python/Cargo.toml

@@ -5,7 +5,7 @@
 //! [`pyo3`]: https://pyo3.rs/
 use std::any::Any;
 use std::ops::{ControlFlow, Range};
-use std::sync::Arc;
+use std::sync::{Arc, Once};
 
 use arrow::datatypes::DataType;
 use datafusion_common::{
@@ -23,6 +23,7 @@ use crate::signature::PythonFn;
 mod conversion;
 mod error;
 mod inspect;
+mod python_modules;
 mod signature;
 
 /// Supported Python version range.
@@ -271,15 +272,20 @@ fn root() -> Option<Vec<u8>> {
 ///
 /// [Python Standard Library]: https://docs.python.org/3/library/index.html
 fn init_python() {
-    Python::initialize();
-
-    Python::attach(|py| {
-        let version_info = py.version_info();
-        let version_tuple = (version_info.major, version_info.minor, version_info.patch);
-        assert!(
-            PYTHON_VERSION_RANGE.contains(&version_tuple),
-            "Unsupported python version: {version_tuple:?}, supported range is {PYTHON_VERSION_RANGE:?}",
-        );
+    static INIT: Once = Once::new();
+
+    INIT.call_once(|| {
+        python_modules::register();
+        Python::initialize();
+
+        Python::attach(|py| {
+            let version_info = py.version_info();
+            let version_tuple = (version_info.major, version_info.minor, version_info.patch);
+            assert!(
+                PYTHON_VERSION_RANGE.contains(&version_tuple),
+                "Unsupported python version: {version_tuple:?}, supported range is {PYTHON_VERSION_RANGE:?}",
+            );
+        });
     });
 }
 

guests/python/src/lib.rs

@@ -0,0 +1,37 @@
+//! Error handling helpers.
+use pyo3::{exceptions::PyValueError, prelude::*};
+
+/// A resource (handle) was already used/moved/closed.
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct ResourceMoved;
+
+impl std::fmt::Display for ResourceMoved {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ResourceMoved")
+    }
+}
+
+impl std::error::Error for ResourceMoved {}
+
+impl From<ResourceMoved> for PyErr {
+    fn from(e: ResourceMoved) -> Self {
+        PyValueError::new_err(e.to_string())
+    }
+}
+
+/// Extensions trait for [`Option`] to simplify the work with [`ResourceMoved`].
+pub(crate) trait ResourceMovedOptionExt {
+    /// Option type.
+    type T;
+
+    /// Require the [`Option`] to be [`Some`], otherwise fail with [`ResourceMoved`].
+    fn require_resource(self) -> Result<Self::T, ResourceMoved>;
+}
+
+impl<T> ResourceMovedOptionExt for Option<T> {
+    type T = T;
+
+    fn require_resource(self) -> Result<Self::T, ResourceMoved> {
+        self.ok_or(ResourceMoved)
+    }
+}