Make sink offset parameter nullable for consistent API

Author: gwaramadze

quixstreams/dataframe/dataframe.py

@@ -1708,7 +1708,7 @@ def _sink_callback(
                 headers=headers,
                 partition=ctx.partition,
                 topic=ctx.topic,
-                offset=ctx.offset or 0,
+                offset=ctx.offset,
             )
 
         # uses apply without returning to make this operation terminal

quixstreams/sinks/base/batch.py

@@ -1,6 +1,6 @@
 from collections import deque
 from itertools import islice
-from typing import Any, Deque, Iterable, Iterator
+from typing import Any, Deque, Iterable, Iterator, Optional
 
 from quixstreams.models import HeadersTuples
 
@@ -39,7 +39,7 @@ def size(self) -> int:
         return len(self._buffer)
 
     @property
-    def start_offset(self) -> int:
+    def start_offset(self) -> Optional[int]:
         return self._buffer[0].offset
 
     def append(
@@ -48,7 +48,7 @@ def append(
         key: Any,
         timestamp: int,
         headers: HeadersTuples,
-        offset: int,
+        offset: Optional[int],
     ):
         self._buffer.append(
             SinkItem(

quixstreams/sinks/base/item.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional
 
 from quixstreams.models import HeadersTuples
 
@@ -20,7 +20,7 @@ def __init__(
         key: Any,
         timestamp: int,
         headers: HeadersTuples,
-        offset: int,
+        offset: Optional[int],
     ):
         self.key = key
         self.value = value

quixstreams/sinks/base/sink.py

@@ -71,7 +71,7 @@ def add(
         headers: HeadersTuples,
         topic: str,
         partition: int,
-        offset: int,
+        offset: Optional[int],
     ):
         """
         This method is triggered on every new processed record being sent to this sink.
@@ -164,7 +164,7 @@ def add(
         headers: HeadersTuples,
         topic: str,
         partition: int,
-        offset: int,
+        offset: Optional[int],
     ):
         """
         Add a new record to in-memory batch.

quixstreams/sinks/core/influxdb3.py

@@ -235,7 +235,7 @@ def add(
         headers: HeadersTuples,
         topic: str,
         partition: int,
-        offset: int,
+        offset: Optional[int],
     ):
         if not isinstance(value, Mapping):
             raise TypeError(

quixstreams/sinks/core/list.py

@@ -1,5 +1,5 @@
 from collections import UserList
-from typing import Any
+from typing import Any, Optional
 
 from quixstreams.models import HeadersTuples
 from quixstreams.sinks.base import BaseSink
@@ -63,7 +63,7 @@ def add(
         headers: HeadersTuples,
         topic: str,
         partition: int,
-        offset: int,
+        offset: Optional[int],
     ):
         if not isinstance(value, dict):
             value = {"value": value}