fix(lake): prefix should be underscore

Author: drgroot

.github/workflows/test.yml

@@ -58,5 +58,5 @@ jobs:
           BUS_URL: amqp://guest:guest@${{ github.server_url != 'https://github.com' && 'rabbitmq' || 'localhost' }}
         run: |
           pip install coverage
-          coverage run -m unittest tests/**/*.py
+          coverage run -m unittest tests/*.py
           coverage report -m --fail-under=60

servc/svc/com/storage/lake.py

@@ -45,7 +45,7 @@ def _get_table_name(self) -> str:
             name_w_medallion = self._table
         else:
             name_w_medallion = "".join(
-                [self._table["medallion"].value, "-", self._table["name"]]
+                [self._table["medallion"].value, "_", self._table["name"]]
             )
 
         return ".".join([schema, name_w_medallion])

servc/svc/com/worker/hooks/parallelize.py

@@ -57,8 +57,7 @@ def evaluate_part_pre_hook(
 
     jobs = resolvers[part_method](message["id"], artifact, context)
     if not isinstance(jobs, list):
-        print(f"Resolver {part_method} did not return a list")
-        return True
+        raise Exception(f"Resolver {part_method} did not return a list")
 
     # formulate on complete hook
     complete_hook: List[OnCompleteHook] = []
@@ -84,8 +83,9 @@ def evaluate_part_pre_hook(
             complete_hook.append(newHook)
 
     # create task queue
-    task_queue = f"part.{route}-{method}-{message['id']}"
-    bus.create_queue(task_queue, False)
+    if len(jobs):
+        task_queue = f"part.{route}-{method}-{message['id']}"
+        bus.create_queue(task_queue, False)
 
     # publish messages to part queue
     payload_template: InputPayload = {

tests/__init__.py

@@ -0,0 +1,23 @@
+import json
+
+
+def get_route_message(channel, cache, route, deleteRoute=False):
+    queue = channel.queue_declare(
+        queue=route,
+        passive=True,
+        durable=True,
+        exclusive=False,
+        auto_delete=False,
+    )
+    count = queue.method.message_count
+    body = None
+
+    if count:
+        _m, _h, body = channel.basic_get(route)
+    if deleteRoute:
+        channel.queue_delete(queue=route)
+    if body:
+        body = json.loads(body.decode("utf-8"))
+        if "argumentId" in body:
+            body["argument"] = cache.getKey(body["argumentId"])
+    return body, count

tests/lake/test_iceberg.py

@@ -48,7 +48,7 @@ def test_connect(self):
         self.assertTrue(self.iceberg.isOpen)
 
     def test_name(self):
-        self.assertEqual(self.iceberg.tablename, "default.bronze-test")
+        self.assertEqual(self.iceberg.tablename, "default.bronze_test")
 
     def test_insert(self):
         self.iceberg.overwrite([])
@@ -114,7 +114,7 @@ def test_load_from_catalog(self):
         self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
         orig_data = self.iceberg.read(["date"]).to_pylist()
 
-        iceberg = IceBerg(config, "default.bronze-test")
+        iceberg = IceBerg(config, "default.bronze_test")
         iceberg._connect()
         self.assertTrue(iceberg.isOpen)
 

tests/test_complete.py

@@ -0,0 +1,83 @@
+import unittest
+
+import pika
+
+from servc.svc.com.bus.rabbitmq import BusRabbitMQ
+from servc.svc.com.cache.redis import CacheRedis
+from servc.svc.com.worker.hooks.oncomplete import process_complete_hook
+from servc.svc.config import Config
+from servc.svc.io.hooks import CompleteHookType
+from servc.svc.io.input import ArgumentArtifact, InputPayload, InputType
+from tests import get_route_message
+
+message: InputPayload = {
+    "id": "123",
+    "type": InputType.INPUT.value,
+    "route": "test",
+    "argumentId": "",
+}
+art: ArgumentArtifact = {
+    "method": "test",
+    "inputs": {"id": "123"},
+    "hooks": {
+        "on_complete": [
+            {
+                "type": CompleteHookType.SENDMESSAGE,
+                "method": "test",
+                "route": "random",
+            }
+        ]
+    },
+}
+
+
+class TestCompleteHook(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        config = Config()
+        cls.bus = BusRabbitMQ(config.get("conf.bus"))
+        cls.cache = CacheRedis(config.get("conf.cache"))
+
+        params = pika.URLParameters(config.get("conf.bus.url"))
+        cls.conn = pika.BlockingConnection(params)
+        cls.channel = cls.conn.channel()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.bus.delete_queue("random")
+        cls.cache.close()
+        cls.bus.close()
+        cls.channel.close()
+        cls.conn.close()
+
+    def setUp(self):
+        self.bus.create_queue("random", False)
+
+    def tearDown(self):
+        self.bus.delete_queue("random")
+
+    def test_complete_hook_simple(self):
+        res = process_complete_hook(
+            self.bus, self.cache, message, art, art["hooks"]["on_complete"][0]
+        )
+        body, _ = get_route_message(self.channel, self.cache, "random")
+
+        self.assertTrue(body["argument"]["inputs"]["inputs"], art["inputs"])
+        self.assertTrue(res)
+
+    def test_w_hook_override(self):
+        hook = {
+            "type": CompleteHookType.SENDMESSAGE,
+            "method": "test",
+            "route": "random",
+            "inputs": True,
+        }
+        res = process_complete_hook(self.bus, self.cache, message, art, hook)
+        body, _ = get_route_message(self.channel, self.cache, "random")
+
+        self.assertTrue(body["argument"]["inputs"], True)
+        self.assertTrue(res)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_config.py

@@ -0,0 +1,36 @@
+import unittest
+
+from servc.svc.config import Config
+
+
+class TestConfig(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.config = Config()
+
+    def test_get_defaults(self):
+        self.assertEqual(self.config.get("conf.file"), "/config/config.yaml")
+        self.assertEqual(self.config.get("conf.bus.routemap"), {})
+        self.assertEqual(self.config.get("conf.bus.prefix"), "")
+
+    def test_value(self):
+        self.config.setValue("conf.bus.prefix", "test")
+        self.assertEqual(self.config.get("conf.bus.prefix"), "test")
+
+        self.config.setValue("conf.bus.routemap.api", "test_route")
+        self.assertEqual(self.config.get("conf.bus.routemap.api"), "test_route")
+
+        self.config.setValue("conf.bus.routemap_DOT_api", "test_route")
+        self.assertEqual(self.config.get("conf.bus.routemap_DOT_api"), "test_route")
+        self.assertIn("routemap.api", self.config.get("conf.bus"))
+
+    def test_wrong_location(self):
+        try:
+            Config("config.test.yaml")
+        except FileNotFoundError:
+            return self.assertTrue(True)
+        self.assertTrue(False)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_iceberg.py

@@ -0,0 +1,167 @@
+import unittest
+
+import pyarrow as pa
+import pyiceberg.types as types
+from pyiceberg.expressions import EqualTo
+from pyiceberg.schema import Schema
+
+from servc.svc.com.storage.iceberg import IceBerg
+from servc.svc.com.storage.lake import LakeTable, Medallion
+
+schema = pa.schema(
+    [
+        ("date", pa.string()),
+        ("some_int", pa.int64()),
+    ]
+)
+
+mytable: LakeTable = {
+    "name": "test",
+    "partitions": ["date"],
+    "medallion": Medallion.BRONZE,
+    "schema": Schema(
+        types.NestedField(field_id=1, name="date", type=types.StringType()),
+        types.NestedField(field_id=2, name="some_int", type=types.IntegerType()),
+    ),
+}
+
+config = {
+    "database": "default",
+    "catalog_name": "default",
+    "catalog_properties": {
+        "type": "sql",
+        "uri": "sqlite:////tmp/pyiceberg.db",
+        "echo": "true",
+        "init_catalog_tables": "true",
+        "warehouse": "file:///tmp/warehouse",
+    },
+}
+
+
+class TestLakeIceberg(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.iceberg = IceBerg(config, mytable)
+
+    def test_connect(self):
+        self.iceberg._connect()
+        self.assertTrue(self.iceberg.isOpen)
+
+    def test_name(self):
+        self.assertEqual(self.iceberg.tablename, "default.bronze_test")
+
+    def test_insert(self):
+        self.iceberg.overwrite([])
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        data = self.iceberg.read(["date"]).to_pylist()
+        self.assertEqual(len(data), 1)
+        self.assertEqual(data, [{"date": "2021-01-01"}])
+
+    def test_overwrite(self):
+        self.iceberg.overwrite([])
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 3}])
+
+        data = self.iceberg.read(["date"]).to_pylist()
+        self.assertEqual(len(data), 3)
+
+        self.iceberg.overwrite([], {"date": ["2021-01-02"]})
+        data = self.iceberg.read(["date"]).to_pylist()
+        self.assertEqual(len(data), 1)
+
+    def test_reading_partitions(self):
+        self.iceberg.overwrite([])
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 3}])
+
+        data = self.iceberg.read(
+            ["date"], partitions={"date": ["2021-01-01"]}
+        ).to_pylist()
+        self.assertEqual(len(data), 1)
+
+        data = self.iceberg.read(
+            ["date"], partitions={"date": ["2021-01-02"]}
+        ).to_pylist()
+        self.assertEqual(len(data), 2)
+
+        data = self.iceberg.read(
+            ["date"], partitions={"date": ["2021-01-02", "2021-01-01"]}
+        ).to_pylist()
+        self.assertEqual(len(data), 3)
+
+        data = self.iceberg.read(
+            ["date"],
+            options={"limit": 1},
+        ).to_pylist()
+        self.assertEqual(len(data), 1)
+
+        data = self.iceberg.read(
+            ["date"],
+            partitions={"date": ["2021-01-02"]},
+            options={"row_filter": EqualTo("some_int", 3)},
+        ).to_pylist()
+        self.assertEqual(len(data), 1)
+
+        data = self.iceberg.read(
+            ["date"],
+            partitions={"date": ["2021-01-02"], "some_int": [3]},
+        ).to_pylist()
+        self.assertEqual(len(data), 1)
+
+    def test_load_from_catalog(self):
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        orig_data = self.iceberg.read(["date"]).to_pylist()
+
+        iceberg = IceBerg(config, "default.bronze_test")
+        iceberg._connect()
+        self.assertTrue(iceberg.isOpen)
+
+        data = iceberg.read(["date"]).to_pylist()
+        self.assertEqual(orig_data, data)
+        self.assertGreater(len(data), 0)
+
+    def test_version_travel(self):
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        orig_data = self.iceberg.read(["date"]).to_pylist()
+        currentVersion = self.iceberg.getCurrentVersion()
+
+        versions = self.iceberg.getVersions()
+        self.assertGreater(len(versions), 0)
+        self.assertIn(currentVersion, versions)
+
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 1}])
+        new_version = self.iceberg.getCurrentVersion()
+        self.assertNotEqual(currentVersion, new_version)
+
+        data = self.iceberg.read(["date"], version=currentVersion).to_pylist()
+        self.assertEqual(len(data), len(orig_data))
+        self.assertEqual(data, orig_data)
+
+    def test_partitions(self):
+        self.iceberg.overwrite([])
+        self.iceberg.insert([{"date": "2021-01-01", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 1}])
+        self.iceberg.insert([{"date": "2021-01-02", "some_int": 3}])
+
+        partitions = self.iceberg.getPartitions()
+        self.assertEqual(list(partitions.keys()), ["date"])
+        self.assertEqual(len(partitions["date"]), 2)
+        self.assertIn("2021-01-01", partitions["date"])
+        self.assertIn("2021-01-02", partitions["date"])
+
+    def test_schema(self):
+        schema = self.iceberg.getSchema()
+        self.assertIsInstance(schema, pa.Schema)
+        self.assertEqual(len(schema.names), 2)
+        self.assertEqual(schema.names, ["date", "some_int"])
+
+    def test_close(self):
+        self.iceberg.close()
+        self.iceberg.connect()
+        self.iceberg.close()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_idgen.py

@@ -0,0 +1,14 @@
+import unittest
+
+from servc.svc.idgen.simple import simple
+
+
+class TestIDGen(unittest.TestCase):
+    def test_simple_idgen(self):
+        route = "/test"
+        message = {"test": "test"}
+        self.assertIsInstance(simple(route, [], message), str)
+
+
+if __name__ == "__main__":
+    unittest.main()