update code

Author: Yuukadesu

.github/workflows/ci.yml

@@ -34,7 +34,7 @@ jobs:
             libsasl2-dev
 
       - name: Cache Cargo
-        uses: Swatinem/rust-cache@2
+        uses: Swatinem/rust-cache@v2
         with:
           shared-key: "linux-build"
 

cli/cli/src/repl.rs

@@ -343,11 +343,7 @@ impl Repl {
             .set_header(
                 vec!["Command", "Usage"]
                     .into_iter()
-                    .map(|s| {
-                        Cell::new(s)
-                            .fg(Color::Cyan)
-                            .add_attribute(Attribute::Bold)
-                    })
+                    .map(|s| Cell::new(s).fg(Color::Cyan).add_attribute(Attribute::Bold))
                     .collect::<Vec<_>>(),
             )
             .add_row(vec!["HELP", "Show this message"])

python/functionstream-api/src/fs_api/context.py

@@ -22,24 +22,24 @@
 
 class Context(abc.ABC):
     """Context object"""
-    
+
     @abc.abstractmethod
     def emit(self, data: bytes, channel: int = 0):
         pass
-    
+
     @abc.abstractmethod
     def emit_watermark(self, watermark: int, channel: int = 0):
         pass
-    
+
     @abc.abstractmethod
     def getOrCreateKVStore(self, name: str) -> KvStore:
         pass
-    
+
     @abc.abstractmethod
     def getConfig(self) -> Dict[str, str]:
         """
         Get global configuration Map
-        
+
         Returns:
             Dict[str, str]: Configuration dictionary
         """

python/functionstream-api/src/fs_api/driver.py

@@ -22,7 +22,7 @@ def init(self, ctx: Context, config: dict):
     @abc.abstractmethod
     def process(self, ctx: Context, source_id: int, data: bytes):
         pass
-    
+
     @abc.abstractmethod
     def process_watermark(self, ctx: Context, source_id: int, watermark: int):
         pass
@@ -34,7 +34,7 @@ def take_checkpoint(self, ctx: Context, checkpoint_id: int):
     @abc.abstractmethod
     def check_heartbeat(self, ctx: Context) -> bool:
         pass
-    
+
     @abc.abstractmethod
     def close(self, ctx: Context):
         pass

python/functionstream-api/src/fs_api/store/iterator.py

@@ -16,11 +16,11 @@
 
 
 class KvIterator(abc.ABC):
-    
+
     @abc.abstractmethod
     def has_next(self) -> bool:
         pass
-    
+
     @abc.abstractmethod
     def next(self) -> Optional[Tuple[bytes, bytes]]:
         pass

python/functionstream-api/src/fs_api/store/store.py

@@ -23,49 +23,49 @@
 
 
 class KvStore(abc.ABC):
-    
+
     @abc.abstractmethod
     def put_state(self, key: bytes, value: bytes):
         pass
-    
+
     @abc.abstractmethod
     def get_state(self, key: bytes) -> Optional[bytes]:
         pass
-    
+
     @abc.abstractmethod
     def delete_state(self, key: bytes):
         pass
-    
+
     @abc.abstractmethod
     def list_states(self, start_inclusive: bytes, end_exclusive: bytes) -> List[bytes]:
         pass
-    
+
     @abc.abstractmethod
     def put(self, key: ComplexKey, value: bytes):
         pass
-    
+
     @abc.abstractmethod
     def get(self, key: ComplexKey) -> Optional[bytes]:
         pass
-    
+
     @abc.abstractmethod
     def delete(self, key: ComplexKey):
         pass
-    
+
     @abc.abstractmethod
     def merge(self, key: ComplexKey, value: bytes):
         pass
-    
+
     @abc.abstractmethod
     def delete_prefix(self, key: ComplexKey):
         pass
-    
+
     @abc.abstractmethod
-    def list_complex(self, key_group: bytes, key: bytes, namespace: bytes, 
+    def list_complex(self, key_group: bytes, key: bytes, namespace: bytes,
                      start_inclusive: bytes, end_exclusive: bytes) -> List[bytes]:
         pass
 
-    
+
     @abc.abstractmethod
     def scan_complex(self, key_group: bytes, key: bytes, namespace: bytes) -> KvIterator:
         pass

python/functionstream-client/src/fs_client/client.py

@@ -503,4 +503,4 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        self.close()
+        self.close()

python/functionstream-client/src/fs_client/exceptions.py

@@ -137,4 +137,4 @@ def _convert_grpc_error(e: grpc.RpcError) -> FsError:
     message = f"{details} (gRPC code: {code.name if code else 'NONE'})"
 
     # 3. Return the instance (caller should raise it)
-    return exception_cls(message, grpc_code=code)
+    return exception_cls(message, grpc_code=code)

python/functionstream-runtime/build.py

@@ -273,4 +273,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    main()

python/functionstream-runtime/src/fs_runtime/runner.py

@@ -32,7 +32,7 @@ def fs_exec(class_name: str, modules: List[Tuple[str, bytes]]) -> None:
     global _DRIVER
 
     try:
-        
+
         # Load all modules in order
         loaded_modules = {}
         for module_name, module_bytes in modules:
@@ -49,19 +49,19 @@ def fs_exec(class_name: str, modules: List[Tuple[str, bytes]]) -> None:
 
             # Create the module
             module = importlib.util.module_from_spec(spec)
-            
+
             # Execute the module source code
             code = compile(module_source, f"<{module_name}>", "exec")
             exec(code, module.__dict__)
-            
+
             # Add the module to sys.modules
             sys.modules[module_name] = module
             loaded_modules[module_name] = module
-        
+
         # Try to find the class in all loaded modules
         ProcessorClass = None
         found_in_module = None
-        
+
         # First, try to find in the last module (most likely location)
         if modules:
             last_module_name = modules[-1][0]
@@ -70,103 +70,103 @@ def fs_exec(class_name: str, modules: List[Tuple[str, bytes]]) -> None:
                 if hasattr(module, class_name):
                     ProcessorClass = getattr(module, class_name)
                     found_in_module = last_module_name
-        
+
         # If not found, search in all modules
         if ProcessorClass is None:
             for module_name, module in loaded_modules.items():
                 if hasattr(module, class_name):
                     ProcessorClass = getattr(module, class_name)
                     found_in_module = module_name
                     break
-        
+
         if ProcessorClass is None:
             module_names = [name for name, _ in modules]
             raise RuntimeError(f"Class '{class_name}' not found in any of the loaded modules: {module_names}")
-        
+
         if not issubclass(ProcessorClass, FSProcessorDriver):
             raise TypeError(f"Class '{class_name}' must be a subclass of FSProcessorDriver")
 
         _DRIVER = ProcessorClass()
-        
+
     except Exception as e:
         raise RuntimeError(f"Failed to load class '{class_name}' from modules: {e}") from e
 
 class WitWorld:
-    
+
     def fs_init(self, config: List[Tuple[str, str]]) -> None:
         global _DRIVER, _CONTEXT
-        
+
         config_dict = convert_config_to_dict(config)
-        
+
         _CONTEXT = WitContext(config_dict)
-        
+
         if _DRIVER:
             try:
                 _DRIVER.init(_CONTEXT, _CONTEXT._CONFIG)
             except Exception:
                 pass
-    
+
     def fs_process(self, source_id: int, data: bytes) -> None:
         global _DRIVER, _CONTEXT
         if not _DRIVER or not _CONTEXT:
             return
-        
+
         try:
             _DRIVER.process(_CONTEXT, source_id, data)
         except Exception:
             pass
-    
+
     def fs_process_watermark(self, source_id: int, watermark: int) -> None:
         global _DRIVER, _CONTEXT
         if not _DRIVER or not _CONTEXT:
             return
-        
+
         try:
             _DRIVER.process_watermark(_CONTEXT, source_id, watermark)
         except Exception:
             pass
-    
+
     def fs_take_checkpoint(self, checkpoint_id: int) -> None:
         global _DRIVER, _CONTEXT
         if not _DRIVER or not _CONTEXT:
             return
-        
+
         try:
             _DRIVER.take_checkpoint(_CONTEXT, checkpoint_id)
         except Exception:
             pass
-    
+
     def fs_check_heartbeat(self) -> bool:
         global _DRIVER, _CONTEXT
         if not _DRIVER or not _CONTEXT:
             return False
-        
+
         try:
             return _DRIVER.check_heartbeat(_CONTEXT)
         except Exception:
             return False
-    
+
     def fs_close(self) -> None:
         global _DRIVER, _CONTEXT
-        
+
         if _DRIVER and _CONTEXT:
             try:
                 _DRIVER.close(_CONTEXT)
             except Exception:
                 pass
-        
+
         _DRIVER = None
         _CONTEXT = None
-    
+
     def fs_exec(self, class_name: str, modules: List[Tuple[str, bytes]]) -> None:
         fs_exec(class_name, modules)
-    
+
     def fs_custom(self, payload: bytes) -> bytes:
         global _DRIVER, _CONTEXT
-        
+
         if not _DRIVER or not _CONTEXT:
             raise RuntimeError("Driver or Context not initialized")
-        
+
         return _DRIVER.custom(payload)