[feat] Create Python bindings for Fluss Admin

bindings/python/fluss/__init__.py

@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from .fluss_python import *
+
+__version__ = "0.1.0"

bindings/python/src/admin.rs

@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use pyo3::prelude::*;
+use pyo3_async_runtimes::tokio::future_into_py;
+use crate::*;
+use std::sync::Arc;
+
+/// Administrative client for managing Fluss tables
+#[pyclass]
+pub struct FlussAdmin {
+    __admin: Arc<fcore::client::FlussAdmin>,
+}
+
+#[pymethods]
+impl FlussAdmin {
+    /// Create a table with the given schema
+    #[pyo3(signature = (table_path, table_descriptor, ignore_if_exists=None))]
+    pub fn create_table<'py>(
+        &self,
+        py: Python<'py>,
+        table_path: &TablePath,
+        table_descriptor: &TableDescriptor,
+        ignore_if_exists: Option<bool>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let ignore = ignore_if_exists.unwrap_or(false);
+
+        let core_table_path = table_path.to_core().clone();
+        let core_descriptor = table_descriptor.to_core().clone();
+        let admin = self.__admin.clone();
+
+        future_into_py(py, async move {
+            admin.create_table(&core_table_path, &core_descriptor, ignore)
+                .await
+                .map_err(|e| FlussError::new_err(e.to_string()))?;
+
+            Python::with_gil(|py| Ok(py.None()))
+        })
+    }
+
+    /// Get table information
+    pub fn get_table<'py>(
+        &self,
+        py: Python<'py>,
+        table_path: &TablePath,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let core_table_path = table_path.to_core().clone();
+        let admin = self.__admin.clone();
+
+        future_into_py(py, async move {
+            let core_table_info = admin.get_table(&core_table_path).await
+                .map_err(|e| FlussError::new_err(format!("Failed to get table: {}", e)))?;
+
+            Python::with_gil(|py| {
+                let table_info = TableInfo::from_core(core_table_info);
+                Py::new(py, table_info)
+            })
+        })
+    }
+
+    /// Get the latest lake snapshot for a table
+    pub fn get_latest_lake_snapshot<'py>(
+        &self,
+        py: Python<'py>,
+        table_path: &TablePath,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let core_table_path = table_path.to_core().clone();
+        let admin = self.__admin.clone();
+
+        future_into_py(py, async move {
+            let core_lake_snapshot = admin.get_latest_lake_snapshot(&core_table_path).await
+                .map_err(|e| FlussError::new_err(format!("Failed to get lake snapshot: {}", e)))?;
+
+            Python::with_gil(|py| {
+                let lake_snapshot = LakeSnapshot::from_core(core_lake_snapshot);
+                Py::new(py, lake_snapshot)
+            })
+        })
+    }
+
+    fn __repr__(&self) -> String {
+        "FlussAdmin()".to_string()
+    }
+}
+
+impl FlussAdmin {
+    // Internal method to create FlussAdmin from core admin
+    pub fn from_core(admin: fcore::client::FlussAdmin) -> Self {
+        Self {
+            __admin: Arc::new(admin),
+        }
+    }
+}

bindings/python/src/connection.rs

@@ -0,0 +1,117 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use pyo3::prelude::*;
+use crate::*;
+use std::sync::Arc;
+use pyo3_async_runtimes::tokio::future_into_py;
+
+/// Connection to a Fluss cluster
+#[pyclass]
+pub struct FlussConnection {
+    inner: Arc<fcore::client::FlussConnection>,
+}
+
+#[pymethods]
+impl FlussConnection {
+    /// Create a new FlussConnection (async)
+    #[staticmethod]
+    fn connect<'py>(py: Python<'py>, config: &Config) -> PyResult<Bound<'py, PyAny>> {
+        let rust_config = config.get_core_config();
+
+        future_into_py(py, async move {
+            let connection = fcore::client::FlussConnection::new(rust_config)
+                .await
+                .map_err(|e| FlussError::new_err(e.to_string()))?;
+
+            let py_connection = FlussConnection {
+                inner: Arc::new(connection),
+            };
+
+            Python::with_gil(|py| {
+                Py::new(py, py_connection)
+            })
+        })
+    }
+
+    /// Get admin interface
+    fn get_admin<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        let client = self.inner.clone();
+
+        future_into_py(py, async move {
+            let admin = client.get_admin()
+                .await
+                .map_err(|e| FlussError::new_err(e.to_string()))?;
+
+            let py_admin = FlussAdmin::from_core(admin);
+
+            Python::with_gil(|py| {
+                Py::new(py, py_admin)
+            })
+        })
+    }
+
+    /// Get a table
+    fn get_table<'py>(&self, py: Python<'py>, table_path: &TablePath) -> PyResult<Bound<'py, PyAny>> {
+        let client = self.inner.clone();
+        let core_path = table_path.to_core().clone();
+
+        future_into_py(py, async move {
+            let core_table = client.get_table(&core_path)
+                .await
+                .map_err(|e| FlussError::new_err(e.to_string()))?;
+
+            let py_table = FlussTable::new_table(
+                client,
+                core_table.metadata,
+                core_table.table_info,
+                core_table.table_path,
+                core_table.has_primary_key,
+            );
+
+            Python::with_gil(|py| {
+                Py::new(py, py_table)
+            })
+        })
+    }
+
+    // Close the connection
+    fn close(&mut self) -> PyResult<()> {
+        Ok(())
+    }
+
+    // Enter the runtime context (for 'with' statement)
+    fn __enter__(slf: PyRef<Self>) -> PyRef<Self> {
+        slf
+    }
+
+    // Exit the runtime context (for 'with' statement)
+    #[pyo3(signature = (_exc_type=None, _exc_value=None, _traceback=None))]
+    fn __exit__(
+        &mut self,
+        _exc_type: Option<PyObject>,
+        _exc_value: Option<PyObject>,
+        _traceback: Option<PyObject>,
+    ) -> PyResult<bool> {
+        self.close()?;
+        Ok(false)
+    }
+
+    fn __repr__(&self) -> String {
+        "FlussConnection()".to_string()
+    }
+}

bindings/python/src/error.rs

@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use pyo3::prelude::*;
+
+/// Fluss errors
+#[pyclass(extends=PyException)]
+#[derive(Debug, Clone)]
+pub struct FlussError {
+    #[pyo3(get)]
+    pub message: String,
+}
+
+#[pymethods]
+impl FlussError {
+    fn __str__(&self) -> String {
+        format!("FlussError: {}", self.message)
+    }
+}
+
+impl FlussError {
+    pub fn new_err(message: impl ToString) -> PyErr {
+        PyErr::new::<FlussError, _>(message.to_string())
+    }
+}

bindings/python/src/lib.rs

@@ -0,0 +1,67 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+pub use ::fluss as fcore;
+use pyo3::prelude::*;
+use once_cell::sync::Lazy;
+use tokio::runtime::Runtime;
+
+mod config;
+mod connection;
+mod table;
+mod admin;
+mod types;
+mod error;
+mod utils;
+
+pub use config::*;
+pub use connection::*;
+pub use table::*;
+pub use admin::*;
+pub use types::*;
+pub use error::*;
+pub use utils::*;
+
+static TOKIO_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
+    tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .expect("Failed to create Tokio runtime")
+});
+
+#[pymodule]
+fn fluss_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    // Register all classes
+    m.add_class::<Config>()?;
+    m.add_class::<FlussConnection>()?;
+    m.add_class::<TablePath>()?;
+    m.add_class::<TableInfo>()?;
+    m.add_class::<TableDescriptor>()?;
+    m.add_class::<FlussAdmin>()?;
+    m.add_class::<FlussTable>()?;
+    m.add_class::<AppendWriter>()?;
+    m.add_class::<Schema>()?;
+    m.add_class::<LogScanner>()?;
+    m.add_class::<LakeSnapshot>()?;
+    m.add_class::<TableBucket>()?;
+
+    // Register exception types
+    // TODO: maybe implement a separate module for exceptions
+    m.add("FlussError", m.py().get_type::<FlussError>())?;
+
+    Ok(())
+}