Merge pull request #67 from serv-c/drgroot-patch-1

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/hooks/__init__.py

@@ -8,7 +8,7 @@
 from servc.svc.config import Config
 from servc.svc.io.hooks import CompleteHookType
 from servc.svc.io.input import ArgumentArtifact, InputPayload, InputType
-from tests.hooks import get_route_message
+from tests import get_route_message
 
 message: InputPayload = {
     "id": "123",

tests/hooks/test_complete.py renamed to tests/test_complete.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)