wip

Author: ovv

quixstreams/app.py

@@ -201,7 +201,6 @@ def __init__(
         :param quix_config_builder: instance of `QuixKafkaConfigsBuilder` to be used
             instead of the default one.
             > NOTE: It is recommended to just use `quix_sdk_token` instead.
-
         """
         configure_logging(loglevel=loglevel)
 

quixstreams/state/metadata.py

@@ -8,3 +8,5 @@
 Undefined = NewType("Undefined", object)
 UNDEFINED = Undefined(object())
 DELETED = Undefined(object())
+
+DEFAULT_PREFIX = b""

quixstreams/state/rocksdb/partition.py

@@ -12,7 +12,6 @@
     int_from_int64_bytes,
     int_to_int64_bytes,
 )
-
 from .exceptions import (
     ColumnFamilyAlreadyExists,
     ColumnFamilyDoesNotExist,

quixstreams/state/rocksdb/transaction.py

@@ -2,17 +2,12 @@
 from typing import Any, Optional
 
 from quixstreams.state.metadata import UNDEFINED
-from quixstreams.state.metadata import (
-    DELETED,
-)
 from quixstreams.state.types import PartitionTransaction
 
-__all__ = ("RocksDBPartitionTransaction", "DEFAULT_PREFIX", "DELETED")
+__all__ = ("RocksDBPartitionTransaction",)
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_PREFIX = b""
-
 
 class RocksDBPartitionTransaction(PartitionTransaction):
     """

quixstreams/state/rocksdb/windowed/transaction.py

@@ -7,12 +7,12 @@
 from quixstreams.state.serialization import (
     serialize,
 )
-from quixstreams.state.types import LoadsFunc, DumpsFunc
+from quixstreams.state.types import LoadsFunc, DumpsFunc, DEFAULT_PREFIX
 
 from .metadata import LATEST_EXPIRED_WINDOW_TIMESTAMP_KEY, LATEST_EXPIRED_WINDOW_CF_NAME
 from .serialization import encode_window_key, encode_window_prefix, parse_window_key
 from .state import WindowedTransactionState
-from ..transaction import RocksDBPartitionTransaction, DEFAULT_PREFIX
+from ..transaction import RocksDBPartitionTransaction
 
 if TYPE_CHECKING:
     from .partition import WindowedRocksDBStorePartition

quixstreams/state/types.py

@@ -29,6 +29,7 @@
     PREFIX_SEPARATOR,
     UNDEFINED,
     Undefined,
+    DEFAULT_PREFIX,
 )
 from quixstreams.state.serialization import (
     serialize,
@@ -499,7 +500,7 @@ def _serialize_key(self, key: Any, prefix: bytes) -> bytes:
         prefix = prefix + PREFIX_SEPARATOR if prefix else b""
         return prefix + key_bytes
 
-    def as_state(self, prefix: Any) -> State:
+    def as_state(self, prefix: Any = DEFAULT_PREFIX) -> State:
         """
         Create an instance implementing the `State` protocol to be provided
         to `StreamingDataFrame` functions.