chore: Datum keys as property (#239)

Signed-off-by: Takashi Menjo <takashi.menjo+github@gmail.com>

Author: tmenjo

pynumaflow/accumulator/_dtypes.py

@@ -76,6 +76,7 @@ def __init__(
         self._headers = headers or {}
         self._id = id_
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event.
 
@@ -488,7 +489,7 @@ def from_datum(cls, datum: Datum):
         """
         return cls(
             value=datum.value,
-            keys=datum.keys(),
+            keys=datum.keys,
             watermark=datum.watermark,
             event_time=datum.event_time,
             headers=datum.headers,

pynumaflow/accumulator/servicer/task_manager.py

@@ -106,7 +106,7 @@ async def close_task(self, req):
         4. Remove the task from the tracker
         """
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys)
         curr_task = self.tasks.get(unified_key, None)
 
@@ -128,7 +128,7 @@ async def create_task(self, req):
         it creates a new task or appends the request to the existing task.
         """
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys)
         curr_task = self.tasks.get(unified_key, None)
 
@@ -179,7 +179,7 @@ async def send_datum_to_task(self, req):
         If the task does not exist, create it.
         """
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys)
         result = self.tasks.get(unified_key, None)
         if not result:

pynumaflow/batchmapper/_dtypes.py

@@ -99,6 +99,7 @@ def __init__(
         self._watermark = watermark
         self._headers = headers or {}
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event"""
         return self._keys

pynumaflow/mapstreamer/_dtypes.py

@@ -151,6 +151,7 @@ def __init__(
         self._watermark = watermark
         self._headers = headers or {}
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event"""
         return self._keys

pynumaflow/reducer/_dtypes.py

@@ -167,6 +167,7 @@ def __init__(
         self._watermark = watermark
         self._headers = headers or {}
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event"""
         return self._keys

pynumaflow/reducer/servicer/task_manager.py

@@ -108,7 +108,7 @@ async def create_task(self, req):
             raise UDFError("reduce create operation error: invalid number of windows")
 
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys, req.windows[0])
         result = self.tasks.get(unified_key, None)
 
@@ -137,7 +137,7 @@ async def append_task(self, req):
         if len(req.windows) != 1:
             raise UDFError("reduce create operation error: invalid number of windows")
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys, req.windows[0])
         result = self.tasks.get(unified_key, None)
         if not result:

pynumaflow/reducestreamer/_dtypes.py

@@ -73,6 +73,7 @@ def __init__(
         self._watermark = watermark
         self._headers = headers or {}
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event"""
         return self._keys

pynumaflow/reducestreamer/servicer/task_manager.py

@@ -110,7 +110,7 @@ async def create_task(self, req):
             raise UDFError("reduce create operation error: invalid number of windows")
 
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys, req.windows[0])
         curr_task = self.tasks.get(unified_key, None)
 
@@ -161,7 +161,7 @@ async def send_datum_to_task(self, req):
         if len(req.windows) != 1:
             raise UDFError("reduce append operation error: invalid number of windows")
         d = req.payload
-        keys = d.keys()
+        keys = d.keys
         unified_key = build_unique_key_name(keys, req.windows[0])
         result = self.tasks.get(unified_key, None)
         if not result:

pynumaflow/sourcetransformer/_dtypes.py

@@ -160,6 +160,7 @@ def __init__(
         self._watermark = watermark
         self._headers = headers or {}
 
+    @property
     def keys(self) -> list[str]:
         """Returns the keys of the event"""
         return self._keys

tests/accumulator/test_async_accumulator.py

@@ -140,15 +140,15 @@ async def handler(self, datums: AsyncIterable[Datum], output: NonBlockingIterato
         async for datum in datums:
             self.counter += 1
             msg = f"counter:{self.counter}"
-            await output.put(Message(str.encode(msg), keys=datum.keys(), tags=[]))
+            await output.put(Message(str.encode(msg), keys=datum.keys, tags=[]))
 
 
 async def accumulator_handler_func(datums: AsyncIterable[Datum], output: NonBlockingIterator):
     counter = 0
     async for datum in datums:
         counter += 1
         msg = f"counter:{counter}"
-        await output.put(Message(str.encode(msg), keys=datum.keys(), tags=[]))
+        await output.put(Message(str.encode(msg), keys=datum.keys, tags=[]))
 
 
 def NewAsyncAccumulator():