add async python interlayer and bump 0.0.3 (#26)

* add async python interlayer

closes #21

Signed-off-by: andylokandy <andylokandy@hotmail.com>

* fix typo in python interlayer

closes #6

Signed-off-by: andylokandy <andylokandy@hotmail.com>

* allow multiple pd endpoints

closes #22

Signed-off-by: andylokandy <andylokandy@hotmail.com>

* remove whitespace

Signed-off-by: andylokandy <andylokandy@hotmail.com>

Author: andylokandy

CHANGELOG

@@ -0,0 +1,15 @@
+# Change Log
+
+* 0.0.3
+
+    * Add `Transaction.rollback()` to abort a transaction.
+
+    * Support multiple PD endpoints.
+
+        Example:
+
+        ```python
+        from tikv_client import RawClient
+
+        client = RawClient.connect(["127.0.0.1:2379"])
+        ```

Cargo.lock

@@ -2,7 +2,7 @@
 authors = ["Andy Lok <andylokandy@hotmail.com>"]
 edition = "2018"
 name = "tikv-client"
-version = "0.0.2"
+version = "0.0.3"
 
 [dependencies]
 futures = "0.3"

Cargo.toml

@@ -1,12 +1,11 @@
 from tikv_client import RawClient
 
-client = RawClient.connect("127.0.0.1:2379")
+client = RawClient.connect(["127.0.0.1:2379"])
 
 client.put(b"k1", b"v1")
 client.put(b"k2", b"v2")
 client.put(b"k3", b"v3")
-client.put(b"k4", b"v4")
-client.put(b"k5", b"v5")
+client.batch_put({b"k4": b"v4", b"k5": b"v5"})
 
 print(client.get(b"k3"))
 print(client.batch_get([b"k1", b"k4"]))

examples/run.py

@@ -2,7 +2,7 @@
 from tikv_client.asynchronous import TransactionClient
 
 async def main():
-    client = await TransactionClient.connect("127.0.0.1:2379")
+    client = await TransactionClient.connect(["127.0.0.1:2379"])
 
     txn = await client.begin(pessimistic=True)
     await txn.put(b"k1", b"v1")

examples/run_async.py

@@ -3,7 +3,6 @@
 use std::convert::TryInto;
 use std::sync::Arc;
 
-use pyo3::exceptions::PyException;
 use pyo3::prelude::*;
 use pyo3::types::*;
 use pyo3::ToPyObject;
@@ -18,17 +17,14 @@ pub struct RawClient {
 
 #[pymethods]
 impl RawClient {
-    #[new]
-    pub fn new() -> PyResult<Self> {
-        Err(PyException::new_err(
-            "Please use `RawClient.connect()` instead.",
-        ))
-    }
-
     #[classmethod]
-    pub fn connect<'p>(_cls: &PyType, py: Python<'p>, pd_endpoint: String) -> PyResult<&'p PyAny> {
+    pub fn connect<'p>(
+        _cls: &PyType,
+        py: Python<'p>,
+        pd_endpoints: Vec<String>,
+    ) -> PyResult<&'p PyAny> {
         future_into_py(py, async move {
-            let inner = tikv_client::RawClient::new(vec![pd_endpoint], None)
+            let inner = tikv_client::RawClient::new(pd_endpoints, None)
                 .await
                 .map_err(to_py_execption)?;
             let client = RawClient {

src/raw.rs

@@ -2,7 +2,6 @@
 
 use std::sync::Arc;
 
-use pyo3::exceptions::PyException;
 use pyo3::prelude::*;
 use pyo3::types::*;
 use pyo3::ToPyObject;
@@ -20,17 +19,14 @@ pub struct TransactionClient {
 
 #[pymethods]
 impl TransactionClient {
-    #[new]
-    pub fn new() -> PyResult<Self> {
-        Err(PyException::new_err(
-            "Please use `TransactionClient.connect()` instead.",
-        ))
-    }
-
     #[classmethod]
-    pub fn connect<'p>(_cls: &PyType, py: Python<'p>, pd_endpoint: String) -> PyResult<&'p PyAny> {
+    pub fn connect<'p>(
+        _cls: &PyType,
+        py: Python<'p>,
+        pd_endpoints: Vec<String>,
+    ) -> PyResult<&'p PyAny> {
         future_into_py(py, async move {
-            let inner = tikv_client::TransactionClient::new(vec![pd_endpoint], None)
+            let inner = tikv_client::TransactionClient::new(pd_endpoints, None)
                 .await
                 .map_err(to_py_execption)?;
             let client = TransactionClient {

src/transaction.rs

@@ -9,10 +9,10 @@ def __init__(self):
         raise Exception("Please use `RawClient.connect()` instead.")
 
     @classmethod
-    def connect(cls, pd_endpoint):
+    def connect(cls, pd_endpoints):
         event_loop = asyncio.get_event_loop()
         inner = event_loop.run_until_complete(
-            asynchronous.RawClient.connect(pd_endpoint))
+            asynchronous.RawClient.connect(pd_endpoints))
         self = cls.__new__(cls)
         self.inner = inner
         return self
@@ -39,7 +39,7 @@ def put(self, key, value, cf="default"):
 
     def batch_put(self, pairs, cf="default"):
         event_loop = asyncio.get_event_loop()
-        event_loop.run_until_complete(self.inner.put(pairs, cf))
+        event_loop.run_until_complete(self.inner.batch_put(pairs, cf))
 
     def delete(self, key, cf="default"):
         event_loop = asyncio.get_event_loop()
@@ -59,10 +59,10 @@ def __init__(self):
         raise Exception("Please use `TransactionClient.connect()` instead.")
 
     @classmethod
-    def connect(cls, pd_endpoint):
+    def connect(cls, pd_endpoints):
         event_loop = asyncio.get_event_loop()
         inner = event_loop.run_until_complete(
-            asynchronous.TransactionClient.connect(pd_endpoint))
+            asynchronous.TransactionClient.connect(pd_endpoints))
         self = cls.__new__(cls)
         self.inner = inner
         return self
@@ -159,3 +159,7 @@ def delete(self, key):
     def commit(self):
         event_loop = asyncio.get_event_loop()
         event_loop.run_until_complete(self.inner.commit())
+
+    def rollback(self):
+        event_loop = asyncio.get_event_loop()
+        event_loop.run_until_complete(self.inner.rollback())

tikv_client/__init__.py

@@ -1,3 +1,133 @@
 # Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
 
-from ..tikv_client import (RawClient, TransactionClient)
+from .. import tikv_client
+
+
+class RawClient:
+    def __init__(self):
+        raise Exception("Please use `RawClient.connect()` instead.")
+
+    @classmethod
+    async def connect(cls, pd_endpoints):
+        if not isinstance(pd_endpoints, list):
+            raise Exception("Please use list as pd_endpoints. For example: `RawClient.connect([127.0.0.1:2379])`.")
+        inner = await tikv_client.RawClient.connect(pd_endpoints)
+        self = cls.__new__(cls)
+        self.inner = inner
+        return self
+
+    async def get(self, key, cf="default"):
+        return await self.inner.get(key, cf)
+
+    async def batch_get(self, keys, cf="default"):
+        return await self.inner.batch_get(keys, cf)
+
+    async def scan(self, start, end, limit, include_start=True, include_end=False, cf="default"):
+        return await self.inner.scan(start, end, limit, include_start, include_end, cf)
+
+    async def scan_keys(self, start, end, limit, include_start=True, include_end=False, cf="default"):
+        return await self.inner.scan_keys(start, end, limit, include_start, include_end, cf)
+
+    async def put(self, key, value, cf="default"):
+        await self.inner.put(key, value, cf)
+
+    async def batch_put(self, pairs, cf="default"):
+        await self.inner.batch_put(pairs, cf)
+
+    async def delete(self, key, cf="default"):
+        await self.inner.delete(key, cf)
+
+    async def batch_delete(self, keys, cf="default"):
+        return await self.inner.batch_delete(keys, cf)
+
+    async def delete_range(self, start, end=None, include_start=True, include_end=False, cf="default"):
+        return await self.inner.delete_range(start, end, include_start, include_end, cf)
+
+
+class TransactionClient:
+    def __init__(self):
+        raise Exception("Please use `TransactionClient.connect()` instead.")
+
+    @classmethod
+    async def connect(cls, pd_endpoints):
+        if not isinstance(pd_endpoints, list):
+            raise Exception("Please use list as pd_endpoints. For example: `TransactionClient.connect([127.0.0.1:2379])`.")
+        inner = await tikv_client.TransactionClient.connect(pd_endpoints)
+        self = cls.__new__(cls)
+        self.inner = inner
+        return self
+
+    async def begin(self, pessimistic=False):
+        transaction = await self.inner.begin(pessimistic)
+        return Transaction(transaction)
+
+    async def current_timestamp(self):
+        return await self.inner.current_timestamp()
+
+    def snapshot(self, timestamp, pessimistic):
+        snapshot = self.inner.snapshot(timestamp, pessimistic)
+        return Snapshot(snapshot)
+
+
+class Snapshot:
+    def __init__(self, inner):
+        self.inner = inner
+
+    async def get(self, key):
+        return await self.inner.get(key)
+
+    async def key_exists(self, key):
+        return await self.inner.key_exists(key)
+
+    async def batch_get(self, keys):
+        return await self.inner.batch_get(keys)
+
+    async def scan(self, start, end, limit, include_start=True, include_end=False):
+        return await self.inner.scan(start, end, limit, include_start, include_end)
+
+    async def scan_keys(self, start, end, limit, include_start=True, include_end=False):
+        return await self.inner.scan_keys(start, end, limit, include_start, include_end)
+
+
+class Transaction:
+    def __init__(self, inner):
+        self.inner = inner
+
+    async def get(self, key):
+        return await self.inner.get(key)
+
+    async def get_for_update(self, key):
+        return await self.inner.get_for_update(key)
+
+    async def key_exists(self, key):
+        return await self.inner.key_exists(key)
+
+    async def batch_get(self, keys):
+        return await self.inner.batch_get(keys)
+
+    async def batch_get_for_update(self, keys):
+        return await self.inner.batch_get_for_update(keys)
+
+    async def scan(self, start, end, limit, include_start=True, include_end=False):
+        return await self.inner.scan(start, end, limit, include_start, include_end)
+
+    async def scan_keys(self, start, end, limit, include_start=True, include_end=False):
+        return await self.inner.scan_keys(start, end, limit, include_start, include_end)
+
+    async def lock_keys(self, keys):
+        await self.inner.lock_keys(keys)
+
+    async def put(self, key, value):
+        await self.inner.put(key, value)
+
+    async def insert(self, key, value):
+        await self.inner.insert(key, value)
+
+    async def delete(self, key):
+        await self.inner.delete(key)
+
+    async def commit(self):
+        await self.inner.commit()
+
+    async def rollback(self):
+        await self.inner.rollback()