WIP - Implement PyTxn

src/communicator.rs

@@ -111,8 +111,8 @@ impl PyCommunicator {
         todo!();
 
         // future_into_py(py, async move {
-            // let mut client = client.lock().await;
-            // result.map(|_| ()).map_err(|e| Error(e).into())
+        // let mut client = client.lock().await;
+        // result.map(|_| ()).map_err(|e| Error(e).into())
         // })
     }
 
@@ -121,8 +121,9 @@ impl PyCommunicator {
         todo!();
 
         // future_into_py(py, async move {
-            // let mut client = client.lock().await;
-            // result.map(|_| ()).map_err(|e| Error(e).into())
+        //     let mut client = client.lock().await;
+        //     client.txn(txn)
+        //     result.map(|_| ()).map_err(|e| Error(e).into())
         // })
     }
 
@@ -131,8 +132,8 @@ impl PyCommunicator {
         todo!();
 
         // future_into_py(py, async move {
-            // let mut client = client.lock().await;
-            // result.map(|_| ()).map_err(|e| Error(e).into())
+        //     let mut client = client.lock().await;
+        //     result.map(|_| ()).map_err(|e| Error(e).into())
         // })
     }
 

src/compare.rs

@@ -0,0 +1,48 @@
+use etcd_client::{Compare, CompareOp};
+use pyo3::prelude::*;
+use pyo3::pyclass::CompareOp as PyO3CompareOp;
+
+// #[derive(Clone)]
+// #[pyclass(name = "CompareOp")]
+// pub struct PyCompareOp(CompareOp);
+
+// #[pymethods]
+// impl PyCompareOp {
+//     #[classattr]
+//     const EQUAL: Self = Self(CompareOp::Equal);
+//     #[classattr]
+//     const GREATER: Self = Self(CompareOp::Greater);
+//     #[classattr]
+//     const LESS: Self = Self(CompareOp::Less);
+//     #[classattr]
+//     const NOT_EQUAL: Self = Self(CompareOp::NotEqual);
+
+//     pub fn __repr__(&self) -> String {
+//         match self.0 {
+//             CompareOp::Equal => "EQUAL".to_owned(),
+//             CompareOp::Greater => "GREATER".to_owned(),
+//             CompareOp::Less => "LESS".to_owned(),
+//             CompareOp::NotEqual => "NOT_EQUAL".to_owned(),
+//         }
+//     }
+
+//     pub fn __richcmp__(&self, py: Python, rhs: &PyCompareOp, op: PyO3CompareOp) -> PyObject {
+//         match op {
+//             PyO3CompareOp::Eq => (self.0 == rhs.0).into_py(py),
+//             PyO3CompareOp::Ne => (self.0 != rhs.0).into_py(py),
+//             _ => py.NotImplemented(),
+//         }
+//     }
+// }
+
+#[derive(Clone)]
+#[pyclass(name = "Compare")]
+pub struct PyCompare(pub Compare);
+
+#[pymethods]
+impl PyCompare {
+    // fn when(&self) {
+    //     self.inner.
+    // }
+
+}

src/lib.rs

@@ -1,20 +1,22 @@
 mod client;
 mod communicator;
+mod compare;
 mod condvar;
 mod error;
 mod event;
+mod request_generator;
 mod stream;
+mod transaction;
 mod utils;
 mod watch;
-mod request_generator;
-mod transaction_action;
 
 use client::PyClient;
 use communicator::PyCommunicator;
 use condvar::PyCondVar;
 use error::ClientError;
 use event::{PyEvent, PyEventType};
 use pyo3::prelude::*;
+use transaction::PyTxn;
 use watch::PyWatch;
 
 #[pymodule]
@@ -25,6 +27,7 @@ fn etcd_client(py: Python, module: &PyModule) -> PyResult<()> {
     module.add_class::<PyEventType>()?;
     module.add_class::<PyWatch>()?;
     module.add_class::<PyCondVar>()?;
+    module.add_class::<PyTxn>()?;
 
     module.add("ClientError", py.get_type::<ClientError>())?;
     Ok(())

src/request_generator.rs

@@ -0,0 +1 @@
+

src/stream.rs

@@ -9,7 +9,7 @@ pub struct PyEventStream {
     stream: WatchStream,
     events: Vec<PyEvent>,
     index: usize,
-    once: bool
+    once: bool,
 }
 
 impl PyEventStream {

src/transaction.rs

@@ -0,0 +1,50 @@
+use etcd_client::{DeleteOptions, GetOptions, PutOptions, Txn, TxnOp};
+use pyo3::prelude::*;
+
+use crate::compare::PyCompare;
+
+#[derive(Clone)]
+#[pyclass(name = "TxnOp")]
+pub struct PyTxnOp(TxnOp);
+
+#[pymethods]
+impl PyTxnOp {
+    #[staticmethod]
+    fn get(key: String) -> PyResult<Self> {
+        let options = GetOptions::new();
+        Ok(PyTxnOp(TxnOp::get(key, Some(options))))
+    }
+    #[staticmethod]
+    fn put(key: String, value: String) -> PyResult<Self> {
+        let options = PutOptions::new();
+        Ok(PyTxnOp(TxnOp::put(key, value, Some(options))))
+    }
+
+    #[staticmethod]
+    fn delete(key: String) -> PyResult<Self> {
+        let options = DeleteOptions::new();
+        Ok(PyTxnOp(TxnOp::delete(key, Some(options))))
+    }
+}
+
+#[derive(Clone)]
+#[pyclass(name = "Txn")]
+pub struct PyTxn(Txn);
+
+#[pymethods]
+impl PyTxn {
+    fn when(&self, compares: Vec<PyCompare>) -> PyResult<Self> {
+        let compares = compares.into_iter().map(|c| c.0).collect::<Vec<_>>();
+        Ok(PyTxn(self.0.clone().when(compares)))
+    }
+
+    fn and_then(&self, operations: Vec<PyTxnOp>) -> PyResult<Self> {
+        let operations = operations.into_iter().map(|c| c.0).collect::<Vec<_>>();
+        Ok(PyTxn(self.0.clone().and_then(operations)))
+    }
+
+    fn or_else(&self, operations: Vec<PyTxnOp>) -> PyResult<Self> {
+        let operations = operations.into_iter().map(|c| c.0).collect::<Vec<_>>();
+        Ok(PyTxn(self.0.clone().or_else(operations)))
+    }
+}

src/watch.rs

@@ -109,7 +109,7 @@ impl PyWatch {
                         }
 
                         result
-                    },
+                    }
                     None => return Err(PyStopAsyncIteration::new_err(())),
                 }?;