Add simple TableView API

Author: BewareMyPower

pulsar/__init__.py

@@ -1126,6 +1126,42 @@ def my_listener(reader, message):
         self._consumers.append(c)
         return c
 
+    def create_table_view(self, topic: str,
+                          subscription_name: Optional[str] = None,
+                          schema: schema.Schema = schema.BytesSchema()) -> TableView:
+        """
+        Create a table view on a particular topic
+
+        Parameters
+        ----------
+
+        topic: str
+            The name of the topic.
+        subscription_name: str, optional
+            The name of the subscription. If it's not specified, a random subscription name
+            will be used.
+        schema: pulsar.schema.Schema, default=pulsar.schema.BytesSchema
+            Define the schema of this table view. If the schema is incompatible with the topic's
+            schema, this method will throw an exception. This schema is also used to deserialize
+            the value of messages in the table view.
+
+        Returns
+        -------
+        TableView
+            A table view instance.
+        """
+        _check_type(str, topic, 'topic')
+        _check_type_or_none(str, subscription_name, 'subscription_name')
+        _check_type(_schema.Schema, schema, 'schema')
+
+        tv_conf = _pulsar.TableViewConfiguration()
+        if subscription_name is not None:
+            tv_conf.subscription_name(subscription_name)
+        tv_conf.schema(schema.schema_info())
+        tv = self._client.create_table_view(topic, tv_conf)
+        self._table_view = TableView(tv, topic, subscription_name, schema)
+        return self._table_view
+
     def get_topic_partitions(self, topic):
         """
         Get the list of partitions for a given topic.

pulsar/tableview.py

@@ -21,7 +21,60 @@
 The TableView implementation.
 """
 
+from typing import Any, Optional
+from pulsar.schema.schema import Schema
+import _pulsar
+
 class TableView():
 
-    def __init__(self) -> None:
-        pass
+    def __init__(self, table_view: _pulsar.TableView, topic: str,
+                 subscription: Optional[str], schema: Schema) -> None:
+        self._table_view = table_view
+        self._topic = topic
+        self._subscription = subscription
+        self._schema = schema
+
+    def get(self, key: str) -> Optional[Any]:
+        """
+        Return the value associated with the given key in the table view.
+
+        Parameters
+        ----------
+        key: str
+            The message key
+
+        Returns
+        -------
+        Optional[Any]
+            The value associated with the key, or None if the key does not exist.
+        """
+        pair = self._table_view.get(key)
+        if pair[0]:
+            return self._schema.decode(pair[1])
+        else:
+            return None
+        #value = self._table_view.get(key)
+        #if value is None:
+        #    return None
+        #return self._schema.decode(value)
+
+    def close(self) -> None:
+        """
+        Close the table view.
+        """
+        self._table_view.close()
+
+    def __len__(self) -> int:
+        """
+        Return the number of entries in the table view.
+        """
+        return self._table_view.size()
+
+    def __str__(self) -> str:
+        if self._subscription is None:
+            return f"TableView(topic={self._topic})"
+        else:
+            return f"TableView(topic={self._topic}, subscription={self._subscription})"
+
+    def __repr__(self) -> str:
+        return self.__str__()

src/table_view.cc

@@ -20,6 +20,9 @@
 #include <pulsar/TableView.h>
 #include <pulsar/Schema.h>
 #include <pulsar/TableViewConfiguration.h>
+#include <pybind11/stl.h>
+#include <utility>
+#include "utils.h"
 
 namespace py = pybind11;
 using namespace pulsar;
@@ -32,5 +35,22 @@ void export_table_view(py::module_& m) {
         .def("schema",
              [](TableViewConfiguration& config, const SchemaInfo& schema) { config.schemaInfo = schema; });
 
-    py::class_<TableView>(m, "TableView").def(py::init<>());
+    py::class_<TableView>(m, "TableView")
+        .def(py::init<>())
+        .def("get",
+             [](const TableView& view, const std::string& key) -> std::pair<bool, py::bytes> {
+                 py::gil_scoped_release release;
+                 std::string value;
+                 bool available = view.getValue(key, value);
+                 py::gil_scoped_acquire acquire;
+                 if (available) {
+                     return std::make_pair(true, py::bytes(std::move(value)));
+                 } else {
+                     return std::make_pair(false, py::bytes());
+                 }
+             })
+        .def("size", &TableView::size, py::call_guard<py::gil_scoped_release>())
+        .def("close", [](TableView& view) {
+            waitForAsyncResult([&view](ResultCallback callback) { view.closeAsync(callback); });
+        });
 }

tests/run-unit-tests.sh

@@ -28,5 +28,6 @@ python3 debug_logger_test.py
 python3 interrupted_test.py
 python3 pulsar_test.py
 python3 schema_test.py
+python3 table_view_test.py
 python3 reader_test.py
 python3 asyncio_test.py

tests/table_view_test.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Callable
+from unittest import TestCase, main
+import time
+
+from pulsar import Client
+
+class TableViewTest(TestCase):
+
+    def setUp(self):
+        self._client: Client = Client('pulsar://localhost:6650')
+
+    def tearDown(self) -> None:
+        self._client.close()
+
+    def test_get(self):
+        topic = f'table_view_test_get-{time.time()}'
+        table_view = self._client.create_table_view(topic)
+        self.assertEqual(len(table_view), 0)
+
+        producer = self._client.create_producer(topic)
+        producer.send('value-0'.encode(), partition_key='key-0')
+        producer.send(b'\xba\xd0\xba\xd0', partition_key='key-1') # an invalid UTF-8 bytes
+
+        self._wait_for_assertion(lambda: self.assertEqual(len(table_view), 2))
+        self.assertEqual(table_view.get('key-0'), b'value-0')
+        self.assertEqual(table_view.get('key-1'), b'\xba\xd0\xba\xd0')
+
+        producer.send('value-1'.encode(), partition_key='key-0')
+        # TODO: Upgrade to C++ client 3.7.1 to include https://github.com/apache/pulsar-client-cpp/pull/487
+        #self._wait_for_assertion(lambda: self.assertEqual(table_view.get('key-0'), b'value-1'))
+
+        producer.close()
+        table_view.close()
+        
+
+    def _wait_for_assertion(self, assertion: Callable, timeout=5) -> None:
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            try:
+                assertion()
+                return
+            except AssertionError:
+                time.sleep(0.1)
+        assertion()
+
+if __name__ == "__main__":
+    main()