WIP

Cargo.toml

@@ -10,14 +10,10 @@ name = "etcd_client"
 crate-type = ["cdylib"]
 
 [dependencies]
-async-recursion = "1.0.5"
 etcd-client = "0.12.4"
-lazy_static = "1.4.0"
 pyo3 = { version = "0.20.2", features = ["extension-module", "multiple-pymethods"] }
 pyo3-asyncio = { version = "0.20.0", features = ["tokio-runtime"] }
 serde = { version = "1.0.195", features = ["derive"] }
 serde_json = "1.0.111"
-thiserror = "1.0.56"
 tokio = { version = "1.32.0", features = ["sync"] }
 tokio-stream = "0.1.14"
-url = "2.5.0"

src/communicator.rs

@@ -2,7 +2,6 @@ use etcd_client::{Client as EtcdClient, PutOptions};
 use etcd_client::{DeleteOptions, GetOptions, WatchOptions};
 use pyo3::exceptions::PyException;
 use pyo3::prelude::*;
-use pyo3::types::PyDict;
 use pyo3_asyncio::tokio::future_into_py;
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -12,7 +11,7 @@ use crate::condvar::PyCondVar;
 use crate::error::Error;
 use crate::transaction::PyTxn;
 use crate::txn_response::PyTxnResponse;
-use crate::PyWatch;
+use crate::watch::PyWatch;
 
 #[pyclass(name = "Communicator")]
 pub struct PyCommunicator(pub Arc<Mutex<EtcdClient>>);
@@ -108,9 +107,7 @@ impl PyCommunicator {
         future_into_py(py, async move {
             let mut client = client.lock().await;
             let result = client.txn(txn.0).await;
-            result
-                .map(|response| PyTxnResponse(response))
-                .map_err(|e| Error(e).into())
+            result.map(PyTxnResponse).map_err(|e| Error(e).into())
         })
     }
 
@@ -127,7 +124,7 @@ impl PyCommunicator {
             match client.get(key.clone(), None).await {
                 Ok(response) => {
                     if let Some(key_value) = response.kvs().get(0) {
-                        if key_value.value_str().unwrap().to_owned() == initial_val {
+                        if *key_value.value_str().unwrap() == initial_val {
                             match client.put(key, new_val, None).await {
                                 Ok(_) => Ok(true), // replace successful
                                 Err(e) => Err(Error(e)),

src/lib.rs

@@ -3,23 +3,22 @@ mod communicator;
 mod compare;
 mod condvar;
 mod error;
-mod event;
-mod request_generator;
 mod stream;
 mod transaction;
 mod txn_response;
 mod watch;
+mod watch_event;
 
 use client::PyClient;
 use communicator::PyCommunicator;
 use compare::{PyCompare, PyCompareOp};
 use condvar::PyCondVar;
 use error::ClientError;
-use event::{PyWatchEvent, PyWatchEventType};
 use pyo3::prelude::*;
 use transaction::{PyTxn, PyTxnOp};
 use txn_response::PyTxnResponse;
 use watch::PyWatch;
+use watch_event::{PyWatchEvent, PyWatchEventType};
 
 #[pymodule]
 fn etcd_client(py: Python, module: &PyModule) -> PyResult<()> {

src/stream.rs

@@ -2,7 +2,7 @@ use etcd_client::WatchStream;
 use pyo3::pyclass;
 use tokio_stream::StreamExt;
 
-use crate::{error::Error, event::PyWatchEvent};
+use crate::{error::Error, watch_event::PyWatchEvent};
 
 #[pyclass(name = "WatchEventStream")]
 pub struct PyWatchEventStream {
@@ -40,7 +40,7 @@ impl PyWatchEventStream {
                     self.events.push(event.clone().into());
                 }
 
-                if self.events.len() > 0 {
+                if !self.events.is_empty() {
                     let event = self.events[self.index].clone();
                     self.index += 1;
                     Some(Ok(event))

src/watch.rs

@@ -78,7 +78,7 @@ impl PyWatch {
 
 #[pymethods]
 impl PyWatch {
-    fn __aiter__<'a>(&'a self) -> Self {
+    fn __aiter__(&self) -> Self {
         self.clone()
     }