fix review remarks

Author: slampy97

tests/coordination/test_coordination_client.py

@@ -1,56 +1,99 @@
+import pytest
+
 import ydb
 
-from ydb.coordination import NodeConfig, ConsistencyMode, RateLimiterCountersMode, CoordinationClient
+from ydb.coordination import (
+    NodeConfig,
+    ConsistencyMode,
+    RateLimiterCountersMode,
+    CoordinationClient,
+    AsyncCoordinationClient,
+)
 
 
 class TestCoordination:
-    def test_coordination_alter_node(self, driver_sync: ydb.Driver):
+    def test_coordination_node_lifecycle(self, driver_sync: ydb.Driver):
         client = CoordinationClient(driver_sync)
-        node_path = "/local/test_alter_node"
+        node_path = "/local/test_node_lifecycle"
 
         try:
             client.delete_node(node_path)
         except ydb.SchemeError:
             pass
 
-        client.create_node(node_path)
+        with pytest.raises(ydb.SchemeError):
+            client.describe_node(node_path)
 
-        new_config = NodeConfig(
-            session_grace_period_millis=12345,
+        initial_config = NodeConfig(
+            session_grace_period_millis=1000,
             attach_consistency_mode=ConsistencyMode.STRICT,
-            read_consistency_mode=ConsistencyMode.RELAXED,
+            read_consistency_mode=ConsistencyMode.STRICT,
             rate_limiter_counters_mode=RateLimiterCountersMode.UNSET,
             self_check_period_millis=0,
         )
+        client.create_node(node_path, initial_config)
 
-        client.alter_node(node_path, new_config)
+        node_descr = client.describe_node(node_path)
+        assert node_descr.path == node_path
+        assert node_descr.config == initial_config
 
-        node_desc = client.describe_node(node_path)
-        node_config = node_desc.config
-        path = node_desc.path
+        updated_config = NodeConfig(
+            session_grace_period_millis=12345,
+            attach_consistency_mode=ConsistencyMode.STRICT,
+            read_consistency_mode=ConsistencyMode.RELAXED,
+            rate_limiter_counters_mode=RateLimiterCountersMode.DETAILED,
+            self_check_period_millis=10,
+        )
+        client.alter_node(node_path, updated_config)
 
-        assert node_path == path
-        assert node_config.session_grace_period_millis == 12345
-        assert node_config.attach_consistency_mode == ConsistencyMode.STRICT
-        assert node_config.read_consistency_mode == ConsistencyMode.RELAXED
+        node_descr = client.describe_node(node_path)
+        assert node_descr.path == node_path
+        assert node_descr.config == updated_config
 
         client.delete_node(node_path)
 
-    def test_coordination_nodes(self, driver_sync: ydb.Driver):
-        client = CoordinationClient(driver_sync)
-        node_path = "/local/test_node"
+        with pytest.raises(ydb.SchemeError):
+            client.describe_node(node_path)
+
+    async def test_coordination_node_lifecycle_async(self, aio_connection):
+        client = AsyncCoordinationClient(aio_connection)
+        node_path = "/local/test_node_lifecycle"
 
         try:
-            client.delete_node(node_path)
+            await client.delete_node(node_path)
         except ydb.SchemeError:
             pass
 
-        client.create_node(node_path)
+        with pytest.raises(ydb.SchemeError):
+            await client.describe_node(node_path)
 
-        node_descr = client.describe_node(node_path)
+        initial_config = NodeConfig(
+            session_grace_period_millis=1000,
+            attach_consistency_mode=ConsistencyMode.STRICT,
+            read_consistency_mode=ConsistencyMode.STRICT,
+            rate_limiter_counters_mode=RateLimiterCountersMode.UNSET,
+            self_check_period_millis=0,
+        )
+        await client.create_node(node_path, initial_config)
+
+        node_descr = await client.describe_node(node_path)
+        assert node_descr.path == node_path
+        assert node_descr.config == initial_config
 
-        node_descr_path = node_descr.path
+        updated_config = NodeConfig(
+            session_grace_period_millis=12345,
+            attach_consistency_mode=ConsistencyMode.STRICT,
+            read_consistency_mode=ConsistencyMode.RELAXED,
+            rate_limiter_counters_mode=RateLimiterCountersMode.DETAILED,
+            self_check_period_millis=10,
+        )
+        await client.alter_node(node_path, updated_config)
 
-        assert node_descr_path == node_path
+        node_descr = await client.describe_node(node_path)
+        assert node_descr.path == node_path
+        assert node_descr.config == updated_config
 
-        client.delete_node(node_path)
+        await client.delete_node(node_path)
+
+        with pytest.raises(ydb.SchemeError):
+            await client.describe_node(node_path)

ydb/_grpc/grpcwrapper/ydb_coordination.py

@@ -1,20 +1,20 @@
 import typing
 from dataclasses import dataclass
 
+from .ydb_coordination_public_types import NodeConfig
 
 if typing.TYPE_CHECKING:
     from ..v4.protos import ydb_coordination_pb2
 else:
     from ..common.protos import ydb_coordination_pb2
 
 from .common_utils import IToProto
-from ydb.coordination import NodeConfig
 
 
 @dataclass
 class CreateNodeRequest(IToProto):
     path: str
-    config: typing.Optional[NodeConfig] = None
+    config: typing.Optional[NodeConfig]
 
     def to_proto(self) -> ydb_coordination_pb2.CreateNodeRequest:
         cfg_proto = self.config.to_proto() if self.config else None
@@ -27,7 +27,7 @@ def to_proto(self) -> ydb_coordination_pb2.CreateNodeRequest:
 @dataclass
 class AlterNodeRequest(IToProto):
     path: str
-    config: typing.Optional[NodeConfig] = None
+    config: NodeConfig
 
     def to_proto(self) -> ydb_coordination_pb2.AlterNodeRequest:
         cfg_proto = self.config.to_proto() if self.config else None

ydb/coordination/__init__.py

@@ -1,11 +1,11 @@
 from .coordination_client import CoordinationClient
+from .coordination_client_async import AsyncCoordinationClient
 
 from ydb._grpc.grpcwrapper.ydb_coordination_public_types import (
     NodeConfig,
     NodeDescription,
     ConsistencyMode,
     RateLimiterCountersMode,
-    CoordinationClientSettings,
 )
 
 __all__ = [
@@ -14,5 +14,5 @@
     "NodeDescription",
     "ConsistencyMode",
     "RateLimiterCountersMode",
-    "CoordinationClientSettings",
+    "AsyncCoordinationClient",
 ]

ydb/coordination/coordination_client.py

@@ -1,100 +1,43 @@
-import typing
 from typing import Optional
 
-from ydb import _apis, issues
-from ydb._grpc.grpcwrapper.ydb_coordination_public_types import NodeConfig, NodeDescription, CoordinationClientSettings
-
-if typing.TYPE_CHECKING:
-    import ydb
-
-
-class CoordinationClient:
-    def __init__(self, driver: "ydb.Driver"):
-        self._driver = driver
-
-    def create_node(
-        self,
-        path: str,
-        config: Optional[NodeConfig] = None,
-        settings: Optional[CoordinationClientSettings] = None,
-    ):
-        proto_config = config.to_proto() if config else None
-        base_driver_settings = settings.to_base_request_settings() if settings is not None else None
-        request = _apis.ydb_coordination.CreateNodeRequest(
-            path=path,
-            config=proto_config,
-        )
-
-        response = self._driver(
-            request,
-            _apis.CoordinationService.Stub,
-            _apis.CoordinationService.CreateNode,
-            settings=base_driver_settings,
+from ydb._grpc.grpcwrapper.ydb_coordination import (
+    CreateNodeRequest,
+    DescribeNodeRequest,
+    AlterNodeRequest,
+    DropNodeRequest,
+)
+from ydb._grpc.grpcwrapper.ydb_coordination_public_types import NodeConfig, NodeDescription
+from ydb.coordination.base_coordination_client import BaseCoordinationClient
+
+
+class CoordinationClient(BaseCoordinationClient):
+    def create_node(self, path: str, config: Optional[NodeConfig], settings=None):
+        return self._call_create(
+            CreateNodeRequest(path=path, config=config).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
         )
-        issues._process_response(response.operation)
-        return response
 
-    def describe_node(
-        self,
-        path: str,
-        settings: Optional[CoordinationClientSettings] = None,
-    ) -> NodeDescription:
-        request = _apis.ydb_coordination.DescribeNodeRequest(path=path)
-        base_driver_settings = settings.to_base_request_settings() if settings is not None else None
-        response = self._driver(
-            request,
-            _apis.CoordinationService.Stub,
-            _apis.CoordinationService.DescribeNode,
-            settings=base_driver_settings,
+    def describe_node(self, path: str, settings=None) -> NodeDescription:
+        return self._call_describe(
+            DescribeNodeRequest(path=path).to_proto(),
+            settings=settings,
+            wrap_args=(self, path),
         )
-        issues._process_response(response.operation)
-
-        result = _apis.ydb_coordination.DescribeNodeResult()
-        response.operation.result.Unpack(result)
-
-        return NodeDescription(
-            path=path,
-            config=NodeConfig.from_proto(result.config),
-        )
-
-    def delete_node(
-        self,
-        path: str,
-        settings: Optional[CoordinationClientSettings] = None,
-    ):
-        base_driver_settings = settings.to_base_request_settings() if settings is not None else None
-        request = _apis.ydb_coordination.DropNodeRequest(path=path)
-        response = self._driver(
-            request,
-            _apis.CoordinationService.Stub,
-            _apis.CoordinationService.DropNode,
-            settings=base_driver_settings,
-        )
-        issues._process_response(response.operation)
-        return response
-
-    def alter_node(
-        self,
-        path: str,
-        new_config: NodeConfig,
-        settings: Optional[CoordinationClientSettings] = None,
-    ):
-        proto_config = new_config.to_proto() if new_config else None
-        base_driver_settings = settings.to_base_request_settings() if settings is not None else None
 
-        request = _apis.ydb_coordination.AlterNodeRequest(
-            path=path,
-            config=proto_config,
+    def alter_node(self, path: str, new_config: NodeConfig, settings=None):
+        return self._call_alter(
+            AlterNodeRequest(path=path, config=new_config).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
         )
 
-        response = self._driver(
-            request,
-            _apis.CoordinationService.Stub,
-            _apis.CoordinationService.AlterNode,
-            settings=base_driver_settings,
+    def delete_node(self, path: str, settings=None):
+        return self._call_delete(
+            DropNodeRequest(path=path).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
         )
-        issues._process_response(response.operation)
-        return response
 
-    def close(self):
-        pass
+    def lock(self):
+        raise NotImplementedError("Will be implemented in future release")

ydb/coordination/coordination_client_async.py

@@ -0,0 +1,43 @@
+from typing import Optional
+
+from ydb._grpc.grpcwrapper.ydb_coordination import (
+    CreateNodeRequest,
+    DescribeNodeRequest,
+    AlterNodeRequest,
+    DropNodeRequest,
+)
+from ydb._grpc.grpcwrapper.ydb_coordination_public_types import NodeConfig, NodeDescription
+from ydb.coordination.base_coordination_client import BaseCoordinationClient
+
+
+class AsyncCoordinationClient(BaseCoordinationClient):
+    async def create_node(self, path: str, config: Optional[NodeConfig] = None, settings=None):
+        return await self._call_create(
+            CreateNodeRequest(path=path, config=config).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
+        )
+
+    async def describe_node(self, path: str, settings=None) -> NodeDescription:
+        return await self._call_describe(
+            DescribeNodeRequest(path=path).to_proto(),
+            settings=settings,
+            wrap_args=(self, path),
+        )
+
+    async def alter_node(self, path: str, new_config: NodeConfig, settings=None):
+        return await self._call_alter(
+            AlterNodeRequest(path=path, config=new_config).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
+        )
+
+    async def delete_node(self, path: str, settings=None):
+        return await self._call_delete(
+            DropNodeRequest(path=path).to_proto(),
+            settings=settings,
+            wrap_args=(self,),
+        )
+
+    async def lock(self):
+        raise NotImplementedError("Will be implemented in future release")