Improve config

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

Author: RobertIndie

sdks/fs-python/examples/config.yaml

@@ -16,7 +16,7 @@
 # This configuration file defines the settings for the string processing function example.
 
 pulsar:
-  service_url: "pulsar://localhost:6650"  # Required: URL of the Pulsar broker
+  serviceUrl: "pulsar://localhost:6650"  # Required: URL of the Pulsar broker
   authPlugin: ""  # Optional: Authentication plugin class name
   authParams: ""  # Optional: Authentication parameters
 
@@ -33,8 +33,8 @@ subscriptionName: "test-sub"
 
 # Optional: List of request source topics
 requestSource:
-  - pulsar:  # SourceSpec structure with pulsar configuration
-      topic: "string-topic"  # Topic name for request messages
+  pulsar:  # SourceSpec structure with pulsar configuration
+    topic: "string-topic"  # Topic name for request messages
 
 # Optional: Output sink configuration
 sink:

sdks/fs-python/fs_sdk/config.py

@@ -1,30 +1,54 @@
 import os
 import yaml
 from typing import Dict, Any, Optional, List
+from pydantic import BaseModel, Field
 
-class Config:
-    def __init__(self, config_path: str = "config.yaml"):
+class PulsarConfig(BaseModel):
+    service_url: str = "pulsar://localhost:6650"
+    auth_plugin: str = ""
+    auth_params: str = ""
+    max_concurrent_requests: int = 10
+    max_producer_cache_size: int = 100
+
+class ModuleConfig(BaseModel):
+    active_module: Optional[str] = None
+    module_configs: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
+
+class SourceSpec(BaseModel):
+    pulsar: Dict[str, Any] = Field(default_factory=dict)
+
+class SinkSpec(BaseModel):
+    pulsar: Dict[str, Any] = Field(default_factory=dict)
+
+class Config(BaseModel):
+    pulsar: PulsarConfig = Field(default_factory=PulsarConfig)
+    module: Optional[str] = None
+    sources: List[SourceSpec] = Field(default_factory=list)
+    requestSource: Optional[SourceSpec] = None
+    sink: Optional[SinkSpec] = None
+    subscription_name: str = "fs-sdk-subscription"
+    name: Optional[str] = None
+    description: Optional[str] = None
+    modules: ModuleConfig = Field(default_factory=ModuleConfig)
+    config: List[Dict[str, Any]] = Field(default_factory=list)
+
+    @classmethod
+    def from_yaml(cls, config_path: str = "config.yaml") -> "Config":
         """
         Initialize configuration from YAML file.
         
         Args:
             config_path (str): Path to the configuration file
-        """
-        self.config_path = config_path
-        self.config = self._load_config()
-        
-    def _load_config(self) -> Dict[str, Any]:
-        """
-        Load configuration from YAML file.
-        
+            
         Returns:
-            Dict[str, Any]: Configuration dictionary
+            Config: Configuration instance
         """
-        if not os.path.exists(self.config_path):
-            raise FileNotFoundError(f"Configuration file not found: {self.config_path}")
+        if not os.path.exists(config_path):
+            raise FileNotFoundError(f"Configuration file not found: {config_path}")
 
-        with open(self.config_path, 'r') as f:
-            return yaml.safe_load(f)
+        with open(config_path, 'r') as f:
+            config_data = yaml.safe_load(f)
+            return cls(**config_data)
 
     def get_config_value(self, config_name: str) -> Any:
         """
@@ -36,93 +60,11 @@ def get_config_value(self, config_name: str) -> Any:
         Returns:
             Any: The configuration value, or None if not found
         """
-        config_section = self.config.get('config', [])
-        for item in config_section:
+        for item in self.config:
             if config_name in item:
                 return item[config_name]
         return None
-            
-    @property
-    def service_url(self) -> str:
-        """Get Pulsar service URL."""
-        return self.config.get('pulsar', {}).get('service_url', 'pulsar://localhost:6650')
-        
-    @property
-    def auth_plugin(self) -> str:
-        """Get Pulsar auth plugin."""
-        return self.config.get('pulsar', {}).get('authPlugin', '')
-        
-    @property
-    def auth_params(self) -> str:
-        """Get Pulsar auth parameters."""
-        return self.config.get('pulsar', {}).get('authParams', '')
-
-    @property
-    def module(self) -> str:
-        """Get the module name."""
-        return self.config.get('module')
-
-    @property
-    def sources(self) -> List[Dict[str, Any]]:
-        """Get the sources configuration."""
-        return self.config.get('sources', [])
-
-    @property
-    def request_sources(self) -> List[Dict[str, Any]]:
-        """Get the request sources configuration."""
-        return self.config.get('requestSource', [])
-
-    @property
-    def sinks(self) -> List[Dict[str, Any]]:
-        """Get the sinks configuration."""
-        return self.config.get('sink', [])
-
-    def get_pulsar_source_config(self, source: Dict[str, Any]) -> Dict[str, Any]:
-        """Get Pulsar source configuration from a source spec."""
-        return source.get('pulsar', {})
-
-    def get_pulsar_sink_config(self, sink: Dict[str, Any]) -> Dict[str, Any]:
-        """Get Pulsar sink configuration from a sink spec."""
-        return sink.get('pulsar', {})
-
-    @property
-    def request_topic(self) -> str:
-        """Get request topic for the active module."""
-        topic = self.config.get('request_topic')
-        if not topic:
-            raise ValueError("request_topic is not set in config.yaml")
-        return topic
-        
-    @property
-    def subscription_name(self) -> str:
-        """Get subscription name for the active module."""
-        return self.config.get('subscription_name', 'fs-sdk-subscription')
-
-    @property
-    def name(self) -> str:
-        """Get the function name."""
-        return self.config.get('name')
-
-    @property
-    def description(self) -> str:
-        """Get the function description."""
-        return self.config.get('description')
-
-    @property
-    def max_concurrent_requests(self) -> int:
-        """Get maximum number of concurrent requests."""
-        return self.config.get('pulsar', {}).get('max_concurrent_requests', 10)
-
-    @property
-    def max_producer_cache_size(self) -> int:
-        """Get maximum number of producers to cache."""
-        return self.config.get('pulsar', {}).get('max_producer_cache_size', 100)
-
-    @property
-    def active_module(self) -> Optional[str]:
-        """Get the name of the active module."""
-        return self.config.get('modules', {}).get('active_module')
 
     def get_module_config(self, module_name: str) -> Dict[str, Any]:
         """Get configuration for a specific module."""
-        return self.config.get('modules', {}).get(module_name, {}) 
+        return self.modules.module_configs.get(module_name, {}) 

sdks/fs-python/fs_sdk/function.py

@@ -57,23 +57,23 @@ def __init__(
             ValueError: If no module is specified in config or if the specified module
                       doesn't have a corresponding process function.
         """
-        self.config = Config(config_path)
+        self.config = Config.from_yaml(config_path)
         self.process_funcs = process_funcs
 
         # Create authentication if specified
         auth = None
-        if self.config.auth_plugin:
+        if self.config.pulsar.auth_plugin:
             auth = pulsar.Authentication(
-                self.config.auth_plugin,
-                self.config.auth_params
+                self.config.pulsar.auth_plugin,
+                self.config.pulsar.auth_params
             )
 
         self.client = Client(
-            self.config.service_url,
+            self.config.pulsar.service_url,
             authentication=auth,
             operation_timeout_seconds=30
         )
-        self.semaphore = asyncio.Semaphore(10)  # Default max concurrent requests
+        self.semaphore = asyncio.Semaphore(self.config.pulsar.max_concurrent_requests)
         self.metrics = Metrics()
         self.metrics_server = MetricsServer(self.metrics)
         self._shutdown_event = asyncio.Event()
@@ -111,21 +111,18 @@ def _setup_consumer(self):
 
         # Collect topics from sources
         for source in self.config.sources:
-            pulsar_config = self.config.get_pulsar_source_config(source)
-            if pulsar_config:
-                topic = pulsar_config.get('topic')
+            if 'pulsar' in source:
+                topic = source['pulsar'].get('topic')
                 if topic:
                     topics.append(topic)
                     logger.info(f"Added source topic: {topic}")
 
         # Collect topics from request sources
-        for source in self.config.request_sources:
-            pulsar_config = self.config.get_pulsar_source_config(source)
-            if pulsar_config:
-                topic = pulsar_config.get('topic')
-                if topic:
-                    topics.append(topic)
-                    logger.info(f"Added request source topic: {topic}")
+        if self.config.requestSource and self.config.requestSource.pulsar:
+            topic = self.config.requestSource.pulsar.get('topic')
+            if topic:
+                topics.append(topic)
+                logger.info(f"Added request source topic: {topic}")
 
         if not topics:
             raise ValueError("No valid sources or request sources found in config")
@@ -255,10 +252,9 @@ async def process_request(self, message):
                     response_topic = message.properties().get('response_topic')
 
                     # If no response_topic is provided, use the sink topic as default
-                    if not response_topic and self.config.sinks:
-                        sink_config = self.config.get_pulsar_sink_config(self.config.sinks[0])
-                        if sink_config and 'topic' in sink_config:
-                            response_topic = sink_config['topic']
+                    if not response_topic and self.config.sink:
+                        if self.config.sink.pulsar and 'topic' in self.config.sink.pulsar:
+                            response_topic = self.config.sink.pulsar['topic']
                             logger.info(f"Using sink topic as default response topic: {response_topic}")
 
                     if not response_topic:

sdks/fs-python/requirements.txt

@@ -1,2 +1,3 @@
 pulsar-client>=3.0.0
-aiohttp>=3.8.0 
+aiohttp>=3.8.0
+pydantic>=2.0.0 

sdks/fs-python/tests/test_function.py

@@ -7,7 +7,7 @@
 import asyncio
 from unittest.mock import Mock, patch, AsyncMock
 from fs_sdk.function import FSFunction
-from fs_sdk.config import Config
+from fs_sdk.config import Config, PulsarConfig, ModuleConfig, SinkSpec, SourceSpec
 from fs_sdk.metrics import Metrics, MetricsServer
 
 class TestFSFunction:
@@ -19,28 +19,17 @@ def mock_config(self):
         config = Mock(spec=Config)
         config.module = "test_module"
         config.subscription_name = "test_subscription"
-        config.service_url = "pulsar://localhost:6650"
-        config.auth_plugin = None
-        config.auth_params = None
-        config.sources = ["source1"]
-        config.request_sources = ["request_source1"]
-        config.sinks = ["sink1"]
-        # Add the .config dict so property methods work
-        config.config = {
-            "pulsar": {
-                "authPlugin": "",
-                "authParams": "",
-                "serviceUrl": "pulsar://localhost:6650"
-            },
-            "module": "test_module",
-            "subscriptionName": "test_subscription",
-            "sources": [],
-            "requestSources": [],
-            "sinks": []
-        }
-        # Mock Pulsar config methods
-        config.get_pulsar_source_config.return_value = {"topic": "test_topic"}
-        config.get_pulsar_sink_config.return_value = {"topic": "response_topic"}
+        config.pulsar = PulsarConfig(
+            service_url="pulsar://localhost:6650",
+            auth_plugin="",
+            auth_params="",
+            max_concurrent_requests=10,
+            max_producer_cache_size=100
+        )
+        config.sources = [SourceSpec(pulsar={"topic": "test_topic"})]
+        config.requestSource = SourceSpec(pulsar={"topic": "request_topic"})
+        config.sink = SinkSpec(pulsar={"topic": "response_topic"})
+        config.modules = ModuleConfig(active_module="test_module")
         return config
 
     @pytest.fixture
@@ -86,7 +75,7 @@ def mock_metrics_server(self):
     def function(self, mock_config, mock_process_func, mock_client, mock_consumer, 
                 mock_producer, mock_metrics, mock_metrics_server):
         """Create a FSFunction instance with mocks, patching Config to avoid file IO."""
-        with patch('fs_sdk.function.Config', return_value=mock_config), \
+        with patch('fs_sdk.function.Config.from_yaml', return_value=mock_config), \
              patch('fs_sdk.function.Client', return_value=mock_client), \
              patch('fs_sdk.function.Metrics', return_value=mock_metrics), \
              patch('fs_sdk.function.MetricsServer', return_value=mock_metrics_server):
@@ -103,7 +92,7 @@ def function(self, mock_config, mock_process_func, mock_client, mock_consumer,
     @pytest.mark.asyncio
     async def test_init(self, mock_config, mock_process_func):
         """Test FSFunction initialization."""
-        with patch('fs_sdk.function.Config', return_value=mock_config), \
+        with patch('fs_sdk.function.Config.from_yaml', return_value=mock_config), \
              patch('fs_sdk.function.Client'), \
              patch('fs_sdk.function.Metrics'), \
              patch('fs_sdk.function.MetricsServer'):
@@ -153,7 +142,7 @@ async def test_process_request_no_response_topic(self, function, mock_metrics):
         message = Mock()
         message.data.return_value = json.dumps({"test": "data"}).encode('utf-8')
         message.properties.return_value = {"request_id": "test_id"}
-        function.config.sinks = []
+        function.config.sink = None
 
         # Execute
         await function.process_request(message)