Fix CI

Signed-off-by: Zike Yang <zike@apache.org>

Author: RobertIndie

sdks/fs-python/README.md

@@ -16,7 +16,9 @@
 
 # FunctionStream Python SDK
 
-FunctionStream SDK is a powerful Python library for building and deploying serverless functions that process messages from Apache Pulsar. It provides a simple yet flexible framework for creating event-driven applications with robust error handling, metrics collection, and resource management.
+FunctionStream SDK is a powerful Python library for building and deploying serverless functions that process messages
+from Apache Pulsar. It provides a simple yet flexible framework for creating event-driven applications with robust error
+handling, metrics collection, and resource management.
 
 ## Features
 
@@ -103,6 +105,7 @@ modules:
 ### FSFunction
 
 The main class for creating serverless functions. It handles:
+
 - Message consumption and processing
 - Response generation
 - Resource management
@@ -112,6 +115,7 @@ The main class for creating serverless functions. It handles:
 ### Configuration
 
 The SDK uses YAML configuration files to define:
+
 - Pulsar connection settings
 - Module selection
 - Topic subscriptions
@@ -121,6 +125,7 @@ The SDK uses YAML configuration files to define:
 ### Metrics
 
 Built-in metrics collection for:
+
 - Request processing time
 - Success/failure rates
 - Message throughput
@@ -138,24 +143,24 @@ Check out the `examples` directory for complete examples:
 ## Best Practices
 
 1. **Error Handling**
-   - Always handle exceptions in your process functions
-   - Use proper logging for debugging
-   - Implement graceful shutdown
+    - Always handle exceptions in your process functions
+    - Use proper logging for debugging
+    - Implement graceful shutdown
 
 2. **Resource Management**
-   - Close resources properly
-   - Use context managers when possible
-   - Monitor resource usage
+    - Close resources properly
+    - Use context managers when possible
+    - Monitor resource usage
 
 3. **Configuration**
-   - Use environment variables for sensitive data
-   - Validate configuration values
-   - Document configuration options
+    - Use environment variables for sensitive data
+    - Validate configuration values
+    - Document configuration options
 
 4. **Testing**
-   - Write unit tests for your functions
-   - Test error scenarios
-   - Validate input/output schemas
+    - Write unit tests for your functions
+    - Test error scenarios
+    - Validate input/output schemas
 
 ## Development
 

sdks/fs-python/examples/Dockerfile

@@ -5,8 +5,8 @@ WORKDIR /function
 COPY requirements.txt .
 RUN pip install -r requirements.txt
 
-COPY string_function.py .
-COPY config.yaml .
-COPY package.yaml .
+COPY main.py .
 
-CMD ["python", "string_function.py"] 
+RUN chmod +x main.py
+
+CMD ["python", "main.py"]

…ks/fs-python/examples/string_function.py sdks/fs-python/examples/main.pysdks/fs-python/examples/string_function.py renamed to sdks/fs-python/examples/main.py sdks/fs-python/examples/string_function.py renamed to sdks/fs-python/examples/main.py

@@ -20,8 +20,7 @@
 import asyncio
 from typing import Dict, Any
 
-from fs_sdk import FSFunction
-from fs_sdk.context import FSContext
+from function_stream import FSFunction, FSContext
 
 async def string_process_function(context: FSContext, data: Dict[str, Any]) -> Dict[str, Any]:
     """

sdks/fs-python/examples/test_string_function.py

@@ -20,7 +20,6 @@
 import pulsar
 import json
 import uuid
-import time
 
 async def test_string_function():
     """

sdks/fs-python/function_stream/__init__.py

@@ -1,27 +1,27 @@
-from .function import FSFunction
-from .module import FSModule
 from .config import Config, PulsarConfig, PulsarSourceConfig, SourceSpec, SinkSpec, Metric
 from .context import FSContext
+from .function import FSFunction
 from .metrics import Metrics, MetricsServer
+from .module import FSModule
 
 __version__ = "0.6.0rc1"
 __all__ = [
     # Core classes
     "FSFunction",
     "FSModule",
-    
+
     # Configuration classes
     "Config",
-    "PulsarConfig", 
+    "PulsarConfig",
     "PulsarSourceConfig",
     "SourceSpec",
     "SinkSpec",
     "Metric",
-    
+
     # Context and utilities
     "FSContext",
-    
+
     # Metrics and monitoring
     "Metrics",
     "MetricsServer"
-] 
+]

sdks/fs-python/function_stream/config.py

@@ -1,8 +1,10 @@
 import os
-import yaml
 from typing import Dict, Any, Optional, List
+
+import yaml
 from pydantic import BaseModel, Field
 
+
 class PulsarConfig(BaseModel):
     """
     Configuration for Pulsar connection settings.
@@ -12,16 +14,17 @@ class PulsarConfig(BaseModel):
     """
     serviceUrl: str = "pulsar://localhost:6650"
     """Pulsar service URL in format 'pulsar://host:port' or 'pulsar+ssl://host:port' for SSL"""
-    
+
     authPlugin: str = ""
     """Authentication plugin class name (e.g., 'org.apache.pulsar.client.impl.auth.AuthenticationTls')"""
-    
+
     authParams: str = ""
     """Authentication parameters in JSON format or key-value pairs"""
-    
+
     max_concurrent_requests: int = 10
     """Maximum number of concurrent requests allowed for this connection"""
 
+
 class PulsarSourceConfig(BaseModel):
     """
     Configuration for Pulsar source/sink specific settings.
@@ -32,6 +35,7 @@ class PulsarSourceConfig(BaseModel):
     topic: str
     """Pulsar topic name to consume from or produce to"""
 
+
 class SourceSpec(BaseModel):
     """
     Specification for data sources.
@@ -42,6 +46,7 @@ class SourceSpec(BaseModel):
     pulsar: Optional[PulsarSourceConfig] = None
     """Pulsar source configuration (optional)"""
 
+
 class SinkSpec(BaseModel):
     """
     Specification for data sinks.
@@ -52,6 +57,7 @@ class SinkSpec(BaseModel):
     pulsar: Optional[PulsarSourceConfig] = None
     """Pulsar sink configuration (optional)"""
 
+
 class Metric(BaseModel):
     """
     Configuration for metrics and monitoring.
@@ -61,6 +67,7 @@ class Metric(BaseModel):
     port: Optional[int] = 9099
     """Port number for metrics endpoint (default: 9099)"""
 
+
 class Config(BaseModel):
     """
     Main configuration class for FunctionStream SDK.
@@ -70,31 +77,31 @@ class Config(BaseModel):
     """
     name: Optional[str] = None
     """Function name identifier (optional)"""
-    
+
     description: Optional[str] = None
     """Function description (optional)"""
-    
+
     pulsar: PulsarConfig = Field(default_factory=PulsarConfig)
     """Pulsar connection configuration"""
-    
+
     module: str = "default"
     """Module name for the function (default: 'default')"""
-    
+
     sources: List[SourceSpec] = Field(default_factory=list)
     """List of input data sources"""
-    
+
     requestSource: Optional[SourceSpec] = None
     """Request source configuration for request-response pattern (optional)"""
-    
+
     sink: Optional[SinkSpec] = None
     """Output sink configuration (optional)"""
-    
+
     subscriptionName: str = "function-stream-sdk-subscription"
     """Pulsar subscription name for consuming messages"""
-    
+
     metric: Metric = Field(default_factory=Metric)
     """Metrics and monitoring configuration"""
-    
+
     config: Dict[str, Any] = Field(default_factory=dict)
     """Custom configuration key-value pairs for function-specific settings"""
 
@@ -118,7 +125,7 @@ def from_yaml(cls, config_path: str = "config.yaml") -> "Config":
         """
         if not os.path.exists(config_path):
             raise FileNotFoundError(f"Configuration file not found: {config_path}")
-            
+
         with open(config_path, 'r') as f:
             config_data = yaml.safe_load(f)
             return cls(**config_data)

sdks/fs-python/function_stream/context.py

@@ -57,4 +57,4 @@ def get_configs(self) -> Dict[str, Any]:
         return self.config.config
 
     def get_module(self) -> str:
-        return self.config.module
+        return self.config.module

sdks/fs-python/function_stream/function.py

@@ -4,23 +4,25 @@
 This module provides the core functionality for creating and managing FunctionStream
 functions. It handles message processing, request/response flow, and resource management.
 """
+import asyncio
 import dataclasses
-import datetime
+import functools
+import inspect
 import json
-import asyncio
 import logging
-import pulsar
+import os
 import time
-import functools
-import inspect
+import typing
+from datetime import datetime, timezone
 from typing import Callable, Any, Dict, Set, Union, Awaitable, get_type_hints, List, Optional
+
+import pulsar
 from pulsar import Client, Producer
+
 from .config import Config
-from .metrics import Metrics, MetricsServer
 from .context import FSContext
+from .metrics import Metrics, MetricsServer
 from .module import FSModule
-import typing
-from datetime import datetime, timezone
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -91,28 +93,28 @@ def is_none_type(annotation):
 
     def is_awaitable_dict(annotation):
         ann = unwrap_annotated(annotation)
-        origin = typing.get_origin(annotation)
+        origin = typing.get_origin(ann)
         args = typing.get_args(ann)
         return origin in (typing.Awaitable,) and len(args) == 1 and is_dict_return(args[0])
 
     def is_awaitable_none(annotation):
         ann = unwrap_annotated(annotation)
-        origin = typing.get_origin(annotation)
+        origin = typing.get_origin(ann)
         args = typing.get_args(ann)
         return origin in (typing.Awaitable,) and len(args) == 1 and is_none_type(args[0])
 
     def is_union_of_dict_and_none(annotation):
         ann = unwrap_annotated(annotation)
-        origin = typing.get_origin(annotation)
-        args = typing.get_args(annotation)
+        origin = typing.get_origin(ann)
+        args = typing.get_args(ann)
         if origin in (typing.Union, Union):
             return (any(is_dict_return(arg) for arg in args) and any(is_none_type(arg) for arg in args))
         return False
 
     def is_awaitable_union_dict_none(annotation):
         ann = unwrap_annotated(annotation)
-        origin = typing.get_origin(annotation)
-        args = typing.get_args(annotation)
+        origin = typing.get_origin(ann)
+        args = typing.get_args(ann)
         if origin in (typing.Awaitable,):
             if len(args) == 1:
                 return is_union_of_dict_and_none(args[0])
@@ -158,8 +160,9 @@ class FSFunction:
     def __init__(
             self,
             process_funcs: Dict[
-                str, Union[Callable[["FSContext", Dict[str, Any]], Union[Dict[str, Any], Awaitable[Dict[str, Any]]]], FSModule]],
-            config_path: str = "config.yaml"
+                str, Union[Callable[
+                    ["FSContext", Dict[str, Any]], Union[Dict[str, Any], Awaitable[Dict[str, Any]]]], FSModule]],
+            config_path: str = None
     ):
         """
         Initialize the FS Function.
@@ -169,13 +172,15 @@ def __init__(
                 Each function must accept two parameters: (context: FSContext, data: Dict[str, Any])
                 and return either a Dict[str, Any] or an Awaitable[Dict[str, Any]].
                 Each module must be an instance of FSModule.
-            config_path (str): Path to the configuration file. Defaults to "config.yaml".
+            config_path (str): Path to the configuration file.
 
         Raises:
             ValueError: If no module is specified in config or if the specified module
                       doesn't have a corresponding process function, or if the function
                       structure is invalid.
         """
+        if config_path is None:
+            config_path = os.getenv("FS_CONFIG_PATH", "config.yaml")
         self.config = Config.from_yaml(config_path)
         self.process_funcs = process_funcs
         self.context = FSContext(self.config)
@@ -217,7 +222,6 @@ def __init__(
         self._tasks_lock = asyncio.Lock()
         self._consumer = None
 
-
         # Create multi-topics consumer
         self._setup_consumer()
 
@@ -257,7 +261,7 @@ def _setup_consumer(self):
             topics,
             subscription_name,
             consumer_type=pulsar.ConsumerType.Shared,
-            unacked_messages_timeout_ms=30_000 # Only for non-ordering guarantee workload
+            unacked_messages_timeout_ms=30_000  # Only for non-ordering guarantee workload
         )
         logger.info(f"Created multi-topics consumer for topics: {topics} with subscription: {subscription_name}")
 
@@ -425,10 +429,12 @@ async def _send_response(self, response_topic: str, request_id: str, msg: List[M
         loop = asyncio.get_event_loop()
         try:
             producer = self._get_producer(response_topic)
+
             def default_serializer(o):
                 if isinstance(o, datetime):
                     return o.isoformat()
                 return str(o)
+
             send_futures = []
             for m in msg:
                 future = loop.create_future()
@@ -442,6 +448,7 @@ def callback(res, msg_id):
                             loop.call_soon_threadsafe(f.set_exception, err)
                         else:
                             loop.call_soon_threadsafe(f.set_result, msg_id)
+
                     return callback
 
                 event_timestamp = None