add sync client

Author: rekby

tests/conftest.py

@@ -99,6 +99,23 @@ async def driver(endpoint, database, event_loop):
     await driver.stop(timeout=10)
 
 
+@pytest.fixture()
+async def driver_sync(endpoint, database, event_loop):
+    driver_config = ydb.DriverConfig(
+        endpoint,
+        database,
+        credentials=ydb.construct_credentials_from_environ(),
+        root_certificates=ydb.load_ydb_root_certificate(),
+    )
+
+    driver = ydb.Driver(driver_config=driver_config)
+    driver.wait(timeout=15)
+
+    yield driver
+
+    driver.stop(timeout=10)
+
+
 @pytest.fixture()
 def topic_consumer():
     return "fixture-consumer"

tests/topics/test_control_plane.py

@@ -38,3 +38,37 @@ async def test_describe_topic(self, driver, topic_path: str, topic_consumer):
                 break
 
         assert has_consumer
+
+
+class TestTopicClientControlPlane:
+    def test_create_topic(self, driver_sync, database):
+        client = driver_sync.topic_client
+
+        topic_path = database + "/my-test-topic"
+
+        client.create_topic(topic_path)
+
+        with pytest.raises(issues.SchemeError):
+            # double create is ok - try create topic with bad path
+            client.create_topic(database)
+
+    def test_drop_topic(self, driver_sync, topic_path):
+        client = driver_sync.topic_client
+
+        client.drop_topic(topic_path)
+
+        with pytest.raises(issues.SchemeError):
+            client.drop_topic(topic_path)
+
+    def test_describe_topic(self, driver_sync, topic_path: str, topic_consumer):
+        res = driver_sync.topic_client.describe(topic_path)
+
+        assert res.self.name == os.path.basename(topic_path)
+
+        has_consumer = False
+        for consumer in res.consumers:
+            if consumer.name == topic_consumer:
+                has_consumer = True
+                break
+
+        assert has_consumer

ydb/aio/driver.py

@@ -4,8 +4,6 @@
 import ydb
 from .. import _utilities
 from ydb.driver import get_config
-from .. import topic
-
 
 def default_credentials(credentials=None):
     if credentials is not None:
@@ -81,6 +79,8 @@ def __init__(
         credentials=None,
         **kwargs
     ):
+        from .. import topic  # local import for prevent cycle import error
+
         config = get_config(
             driver_config,
             connection_string,

ydb/driver.py

@@ -231,6 +231,8 @@ def __init__(
         :param database: A database path
         :param credentials: A credentials. If not specifed credentials constructed by default.
         """
+        from . import topic  # local import for prevent cycle import error
+
         driver_config = get_config(
             driver_config,
             connection_string,
@@ -246,3 +248,4 @@ def __init__(
 
         self.scheme_client = scheme.SchemeClient(self)
         self.table_client = table.TableClient(self, driver_config.table_client_settings)
+        self.topic_client = topic.TopicClient(self, driver_config.topic_client_settings)

ydb/topic.py

@@ -5,6 +5,7 @@
 from . import aio, Credentials, _apis
 
 from . import scheme
+from . import driver
 
 from ._grpc.grpcwrapper.ydb_topic_public_types import (
     DropTopicRequestParams as _DropTopicRequestParams,
@@ -43,6 +44,14 @@
 
 from ._grpc.grpcwrapper import ydb_topic as _ydb_topic
 from ._grpc.grpcwrapper import ydb_topic_public_types as _ydb_topic_public_types
+from ._grpc.grpcwrapper.ydb_topic_public_types import (
+    PublicDescribeTopicResult as TopicDescription,
+    PublicMultipleWindowsStat as TopicStatWindow,
+    PublicPartitionStats as TopicPartitionStats,
+    PublicCodec as TopicCodec,
+    PublicConsumer as TopicConsumer,
+    PublicMeteringMode as TopicMeteringMode,
+)
 
 
 class TopicClientAsyncIO:
@@ -94,7 +103,7 @@ async def create_topic(
             _wrap_operation,
         )
 
-    async def describe(self, path: str, include_stats: bool = False):
+    async def describe(self, path: str, include_stats: bool = False) -> TopicDescription:
         args = locals().copy()
         del args["self"]
         req = _DescribeTopicRequestParams(**args)
@@ -164,8 +173,74 @@ def topic_writer(
 
 
 class TopicClient:
-    def __init__(self, driver, topic_client_settings: "TopicClientSettings" = None):
-        pass
+    _driver: driver.Driver
+    _credentials: Union[Credentials, None]
+
+    def __init__(self, driver: driver.Driver, topic_client_settings: "TopicClientSettings" = None):
+        self._driver = driver
+
+    def create_topic(
+        self,
+        path: str,
+        min_active_partitions: Optional[
+            int
+        ] = None,  # Minimum partition count auto merge would stop working at.
+        partition_count_limit: Optional[
+            int
+        ] = None,  # Limit for total partition count, including active (open for write) and read-only partitions.
+        retention_period: Optional[
+            datetime.timedelta
+        ] = None,  # How long data in partition should be stored
+        retention_storage_mb: Optional[
+            int
+        ] = None,  # How much data in partition should be stored
+        # List of allowed codecs for writers.
+        # Writes with codec not from this list are forbidden.
+        supported_codecs: Optional[List[Union[TopicCodec, int]]] = None,
+        partition_write_speed_bytes_per_second: Optional[
+            int
+        ] = None,  # Partition write speed in bytes per second
+        partition_write_burst_bytes: Optional[
+            int
+        ] = None,  # Burst size for write in partition, in bytes
+        # User and server attributes of topic. Server attributes starts from "_" and will be validated by server.
+        attributes: Optional[Dict[str, str]] = None,
+        # List of consumers for this topic
+        consumers: Optional[List[Union[TopicConsumer, str]]] = None,
+        # Metering mode for the topic in a serverless database
+        metering_mode: Optional[TopicMeteringMode] = None,
+    ):
+        args = locals().copy()
+        del args["self"]
+        req = _ydb_topic_public_types.CreateTopicRequestParams(**args)
+        req = _ydb_topic.CreateTopicRequest.from_public(req)
+        self._driver(
+            req.to_proto(),
+            _apis.TopicService.Stub,
+            _apis.TopicService.CreateTopic,
+            _wrap_operation,
+        )
+
+    def describe(self, path: str, include_stats: bool = False) -> TopicDescription:
+        args = locals().copy()
+        del args["self"]
+        req = _DescribeTopicRequestParams(**args)
+        res = self._driver(
+            req.to_proto(),
+            _apis.TopicService.Stub,
+            _apis.TopicService.DescribeTopic,
+            _create_result_wrapper(_ydb_topic.DescribeTopicResult),
+        )  # type: _ydb_topic.DescribeTopicResult
+        return res.to_public()
+
+    def drop_topic(self, path: str):
+        req = _DropTopicRequestParams(path=path)
+        self._driver(
+            req.to_proto(),
+            _apis.TopicService.Stub,
+            _apis.TopicService.DropTopic,
+            _wrap_operation,
+        )
 
     def topic_reader(
         self,