add topic control plane async

Author: rekby

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Add control plane operations for topic api: create, drop
+
 ## 3.0.1b4 ##
 * Initial implementation of topic reader
 

tests/conftest.py

@@ -4,7 +4,7 @@
 import pytest
 import ydb
 import time
-import subprocess
+from ydb import issues
 
 
 @pytest.fixture(autouse=True, scope="session")
@@ -105,30 +105,21 @@ def topic_consumer():
 
 
 @pytest.fixture()
-def topic_path(endpoint, topic_consumer) -> str:
-    subprocess.run(
-        """docker-compose exec -T ydb /ydb -e grpc://%s -d /local topic drop /local/test-topic"""
-        % endpoint,
-        shell=True,
-        capture_output=True,
-    )
-    res = subprocess.run(
-        """docker-compose exec -T ydb /ydb -e grpc://%s -d /local topic create /local/test-topic"""
-        % endpoint,
-        shell=True,
-        capture_output=True,
-    )
-    assert res.returncode == 0, res.stderr + res.stdout
+@pytest.mark.asyncio()
+async def topic_path(driver, topic_consumer, database) -> str:
+    topic_path = database + "/test-topic"
+
+    try:
+        await driver.topic_client.drop_topic(topic_path)
+    except issues.SchemeError:
+        pass
 
-    res = subprocess.run(
-        """docker-compose exec -T ydb /ydb -e grpc://%s -d /local topic consumer add --consumer %s /local/test-topic"""
-        % (endpoint, topic_consumer),
-        shell=True,
-        capture_output=True,
+    await driver.topic_client.create_topic(
+        path=topic_path,
+        consumers=[topic_consumer],
     )
-    assert res.returncode == 0, res.stderr + res.stdout
 
-    return "/local/test-topic"
+    return topic_path
 
 
 @pytest.fixture()

tests/topics/test_control_plane.py

@@ -0,0 +1,40 @@
+import os.path
+
+import pytest
+
+from ydb import issues
+
+
+@pytest.mark.asyncio
+class TestTopicClientControlPlaneAsyncIO:
+    async def test_create_topic(self, driver, database):
+        client = driver.topic_client
+
+        topic_path = database + "/my-test-topic"
+
+        await client.create_topic(topic_path)
+
+        with pytest.raises(issues.SchemeError):
+            # double create is ok - try create topic with bad path
+            await client.create_topic(database)
+
+    async def test_drop_topic(self, driver, topic_path):
+        client = driver.topic_client
+
+        await client.drop_topic(topic_path)
+
+        with pytest.raises(issues.SchemeError):
+            await client.drop_topic(topic_path)
+
+    async def test_describe_topic(self, driver, topic_path: str, topic_consumer):
+        res = await driver.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/_apis.py

@@ -103,5 +103,8 @@ class TableService(object):
 class TopicService(object):
     Stub = ydb_topic_v1_pb2_grpc.TopicServiceStub
 
+    CreateTopic = "CreateTopic"
+    DescribeTopic = "DescribeTopic"
+    DropTopic = "DropTopic"
     StreamRead = "StreamRead"
     StreamWrite = "StreamWrite"

ydb/_grpc/grpcwrapper/common_utils.py

@@ -1,10 +1,25 @@
+from __future__ import annotations
+
 import abc
 import asyncio
+import datetime
 import typing
+from typing import (
+    Optional,
+    Any,
+    Iterator,
+    AsyncIterator,
+    Callable,
+    Iterable,
+    Union,
+    Coroutine,
+)
 from dataclasses import dataclass
 
 import grpc
 from google.protobuf.message import Message
+from google.protobuf.duration_pb2 import Duration as ProtoDuration
+from google.protobuf.timestamp_pb2 import Timestamp as ProtoTimeStamp
 
 import ydb.aio
 
@@ -21,14 +36,35 @@
 class IFromProto(abc.ABC):
     @staticmethod
     @abc.abstractmethod
-    def from_proto(msg: Message) -> typing.Any:
-        pass
+    def from_proto(msg: Message) -> Any:
+        ...
+
+
+class IFromProtoWithProtoType(IFromProto):
+    @staticmethod
+    @abc.abstractmethod
+    def empty_proto_message() -> Message:
+        ...
 
 
 class IToProto(abc.ABC):
     @abc.abstractmethod
     def to_proto(self) -> Message:
-        pass
+        ...
+
+
+class IFromPublic(abc.ABC):
+
+    @staticmethod
+    @abc.abstractmethod
+    def from_public(o: typing.Any) -> typing.Any:
+        ...
+
+
+class IToPublic(abc.ABC):
+    @abc.abstractmethod
+    def to_public(self) -> typing.Any:
+        ...
 
 
 class UnknownGrpcMessageError(issues.Error):
@@ -76,7 +112,7 @@ def __next__(self):
 
 
 class SyncIteratorToAsyncIterator:
-    def __init__(self, sync_iterator: typing.Iterator):
+    def __init__(self, sync_iterator: Iterator):
         self._sync_iterator = sync_iterator
 
     def __aiter__(self):
@@ -92,21 +128,21 @@ async def __anext__(self):
 
 class IGrpcWrapperAsyncIO(abc.ABC):
     @abc.abstractmethod
-    async def receive(self) -> typing.Any:
+    async def receive(self) -> Any:
         ...
 
     @abc.abstractmethod
     def write(self, wrap_message: IToProto):
         ...
 
 
-SupportedDriverType = typing.Union[ydb.Driver, ydb.aio.Driver]
+SupportedDriverType = Union[ydb.Driver, ydb.aio.Driver]
 
 
 class GrpcWrapperAsyncIO(IGrpcWrapperAsyncIO):
     from_client_grpc: asyncio.Queue
-    from_server_grpc: typing.AsyncIterator
-    convert_server_grpc_to_wrapper: typing.Callable[[typing.Any], typing.Any]
+    from_server_grpc: AsyncIterator
+    convert_server_grpc_to_wrapper: Callable[[Any], Any]
     _connection_state: str
 
     def __init__(self, convert_server_grpc_to_wrapper):
@@ -140,7 +176,7 @@ async def _start_sync_driver(self, driver: ydb.Driver, stub, method):
         )
         self.from_server_grpc = SyncIteratorToAsyncIterator(stream_call.__iter__())
 
-    async def receive(self) -> typing.Any:
+    async def receive(self) -> Any:
         # todo handle grpc exceptions and convert it to internal exceptions
         try:
             grpc_message = await self.from_server_grpc.__anext__()
@@ -168,7 +204,7 @@ class ServerStatus(IFromProto):
     def __init__(
         self,
         status: issues.StatusCode,
-        issues: typing.Iterable[typing.Any],
+        issues: Iterable[Any],
     ):
         self.status = status
         self.issues = issues
@@ -178,7 +214,7 @@ def __str__(self):
 
     @staticmethod
     def from_proto(
-        msg: typing.Union[
+        msg: Union[
             ydb_topic_pb2.StreamReadMessage.FromServer,
             ydb_topic_pb2.StreamWriteMessage.FromServer,
         ]
@@ -198,11 +234,37 @@ def issue_to_str(cls, issue: ydb_issue_message_pb2.IssueMessage):
 
 
 def callback_from_asyncio(
-    callback: typing.Union[typing.Callable, typing.Coroutine]
+    callback: Union[Callable, Coroutine]
 ) -> [asyncio.Future, asyncio.Task]:
     loop = asyncio.get_running_loop()
 
     if asyncio.iscoroutinefunction(callback):
         return loop.create_task(callback())
     else:
         return loop.run_in_executor(None, callback)
+
+
+def proto_duration_from_timedelta(t: Optional[datetime.timedelta]) -> ProtoDuration:
+    if t is None:
+        return None
+    res = ProtoDuration()
+    res.FromTimedelta(t)
+
+
+def proto_timestamp_from_datetime(t: Optional[datetime.datetime]) -> ProtoTimeStamp:
+    if t is None:
+        return None
+
+    res = ProtoTimeStamp()
+    res.FromDatetime(t)
+
+
+def datetime_from_proto_timestamp(ts: Optional[ProtoTimeStamp]) -> Optional[datetime.datetime]:
+    if ts is None:
+        return None
+    return ts.ToDatetime()
+
+def timedelta_from_proto_duration(d: Optional[ProtoDuration]) -> Optional[datetime.timedelta]:
+    if d is None:
+        return None
+    return d.ToTimedelta()

ydb/_grpc/grpcwrapper/ydb_scheme.py

@@ -0,0 +1,36 @@
+import datetime
+import enum
+from dataclasses import dataclass
+from typing import List
+
+
+@dataclass
+class Entry:
+    name: str
+    owner: str
+    type: "Entry.Type"
+    effective_permissions: "Permissions"
+    permissions: "Permissions"
+    size_bytes: int
+    created_at: datetime.datetime
+
+    class Type(enum.IntEnum):
+        UNSPECIFIED = 0
+        DIRECTORY = 1
+        TABLE = 2
+        PERS_QUEUE_GROUP = 3
+        DATABASE = 4
+        RTMR_VOLUME = 5
+        BLOCK_STORE_VOLUME = 6
+        COORDINATION_NODE = 7
+        COLUMN_STORE = 12
+        COLUMN_TABLE = 13
+        SEQUENCE = 15
+        REPLICATION = 16
+        TOPIC = 17
+
+
+@dataclass
+class Permissions:
+    subject: str
+    permission_names: List[str]