fix: convert headers to dict before exposing to users (numaproj#151)

Signed-off-by: Yashash H L <[email protected]>
kohlisid · Apr 10, 2024 · af095a1 · af095a1
1 parent 51a1b37
commit af095a1
Show file tree

Hide file tree

Showing 15 changed files with 27 additions and 19 deletions.
diff --git a/pynumaflow/mapper/servicer/async_servicer.py b/pynumaflow/mapper/servicer/async_servicer.py
@@ -38,7 +38,7 @@ async def MapFn(
                     value=request.value,
                     event_time=request.event_time.ToDatetime(),
                     watermark=request.watermark.ToDatetime(),
-                    headers=request.headers,
+                    headers=dict(request.headers),
                 ),
             )
         except Exception as e:

diff --git a/pynumaflow/mapstreamer/servicer/async_servicer.py b/pynumaflow/mapstreamer/servicer/async_servicer.py
@@ -40,7 +40,7 @@ async def MapStreamFn(
                 value=request.value,
                 event_time=request.event_time.ToDatetime(),
                 watermark=request.watermark.ToDatetime(),
-                headers=request.headers,
+                headers=dict(request.headers),
             ),
         ):
             yield mapstream_pb2.MapStreamResponse(result=res)

diff --git a/pynumaflow/proto/mapper/map_pb2.py b/pynumaflow/proto/mapper/map_pb2.py
diff --git a/pynumaflow/proto/mapstreamer/mapstream_pb2.py b/pynumaflow/proto/mapstreamer/mapstream_pb2.py
diff --git a/pynumaflow/proto/reducer/reduce_pb2.py b/pynumaflow/proto/reducer/reduce_pb2.py
diff --git a/pynumaflow/proto/sideinput/sideinput_pb2.py b/pynumaflow/proto/sideinput/sideinput_pb2.py
diff --git a/pynumaflow/proto/sinker/sink_pb2.py b/pynumaflow/proto/sinker/sink_pb2.py
diff --git a/pynumaflow/proto/sourcer/source_pb2.py b/pynumaflow/proto/sourcer/source_pb2.py
diff --git a/pynumaflow/proto/sourcetransformer/transform_pb2.py b/pynumaflow/proto/sourcetransformer/transform_pb2.py
diff --git a/pynumaflow/reducer/servicer/async_servicer.py b/pynumaflow/reducer/servicer/async_servicer.py
@@ -35,7 +35,7 @@ async def datum_generator(
                 value=d.payload.value,
                 event_time=d.payload.event_time.ToDatetime(),
                 watermark=d.payload.watermark.ToDatetime(),
-                headers=d.payload.headers,
+                headers=dict(d.payload.headers),
             ),
         )
         yield reduce_request

diff --git a/pynumaflow/reducestreamer/servicer/async_servicer.py b/pynumaflow/reducestreamer/servicer/async_servicer.py
@@ -32,6 +32,7 @@ async def datum_generator(
                 value=d.payload.value,
                 event_time=d.payload.event_time.ToDatetime(),
                 watermark=d.payload.watermark.ToDatetime(),
+                headers=dict(d.payload.headers),
             ),
         )
         yield reduce_request

diff --git a/pynumaflow/sinker/servicer/async_servicer.py b/pynumaflow/sinker/servicer/async_servicer.py
@@ -19,7 +19,7 @@ async def datum_generator(
             value=d.value,
             event_time=d.event_time.ToDatetime(),
             watermark=d.watermark.ToDatetime(),
-            headers=d.headers,
+            headers=dict(d.headers),
         )
         yield datum
 

diff --git a/pynumaflow/sinker/servicer/sync_servicer.py b/pynumaflow/sinker/servicer/sync_servicer.py
@@ -16,7 +16,7 @@ def datum_generator(request_iterator: Iterable[sink_pb2.SinkRequest]) -> Iterabl
             value=d.value,
             event_time=d.event_time.ToDatetime(),
             watermark=d.watermark.ToDatetime(),
-            headers=d.headers,
+            headers=dict(d.headers),
         )
         yield datum
 

diff --git a/pynumaflow/sourcer/servicer/sync_servicer.py b/pynumaflow/sourcer/servicer/sync_servicer.py
@@ -63,7 +63,7 @@ def __invoke_source_read_stream(self, req: ReadRequest):
                     keys=msg.keys,
                     offset=msg.offset.as_dict,
                     event_time=event_time_timestamp,
-                    headers=msg.headers,
+                    headers=dict(msg.headers),
                 )
         except Exception as err:
             _LOGGER.critical("User-Defined Source ReadError ", exc_info=True)

diff --git a/pynumaflow/sourcetransformer/servicer/server.py b/pynumaflow/sourcetransformer/servicer/server.py
@@ -41,7 +41,7 @@ def SourceTransformFn(
                     value=request.value,
                     event_time=request.event_time.ToDatetime(),
                     watermark=request.watermark.ToDatetime(),
-                    headers=request.headers,
+                    headers=dict(request.headers),
                 ),
             )
         except Exception as err: