update

Author: Tom Softreck

src/dialogchain/engine/connector.py

@@ -4,15 +4,11 @@
 This module handles the creation and management of source and destination connectors.
 """
 
-from typing import Dict, Any, Optional, Type, Union
-import logging
-import importlib
-from urllib.parse import urlparse, parse_qs
+from typing import Dict, Any, Optional, Type, Union, List
+from urllib.parse import urlparse, parse_qs, ParseResult
 
 from ..connectors import Source, Destination, ConnectorError
 
-logger = logging.getLogger(__name__)
-
 class ConnectorManager:
     """Manages the creation and lifecycle of connectors."""
 
@@ -63,7 +59,6 @@ def register_source(self, scheme: str, source_class: Type[Source]) -> None:
             source_class: Source class to register
         """
         self.source_types[scheme] = source_class
-        logger.debug(f"Registered source connector: {scheme} -> {source_class.__name__}")
 
     def register_destination(self, scheme: str, dest_class: Type[Destination]) -> None:
         """Register a destination connector class.
@@ -73,7 +68,6 @@ def register_destination(self, scheme: str, dest_class: Type[Destination]) -> No
             dest_class: Destination class to register
         """
         self.destination_types[scheme] = dest_class
-        logger.debug(f"Registered destination connector: {scheme} -> {dest_class.__name__}")
 
     def create_source(self, config: Union[str, Dict[str, Any]]) -> Source:
         """Create a source connector from a URI or config dictionary.
@@ -90,10 +84,9 @@ def create_source(self, config: Union[str, Dict[str, Any]]) -> Source:
         """
         if isinstance(config, str):
             return self._create_source_from_uri(config)
-        elif isinstance(config, dict):
+        if isinstance(config, dict):
             return self._create_source_from_config(config)
-        else:
-            raise ValueError(f"Invalid source config type: {type(config)}")
+        raise ValueError(f"Invalid source config type: {type(config)}")
 
     def create_destination(self, config: Union[str, Dict[str, Any]]) -> Destination:
         """Create a destination connector from a URI or config dictionary.
@@ -110,171 +103,98 @@ def create_destination(self, config: Union[str, Dict[str, Any]]) -> Destination:
         """
         if isinstance(config, str):
             return self._create_destination_from_uri(config)
-        elif isinstance(config, dict):
+        if isinstance(config, dict):
             return self._create_destination_from_config(config)
-        else:
-            raise ValueError(f"Invalid destination config type: {type(config)}")
+        raise ValueError(f"Invalid destination config type: {type(config)}")
 
     def _create_source_from_uri(self, uri: str) -> Source:
-        """Create a source connector from a URI.
-        
-        Args:
-            uri: Source URI (e.g., 'timer:5s', 'imap://user:pass@server')
-            
-        Returns:
-            Configured Source instance
-        """
-        if '://' not in uri and ':' in uri:
-            # Handle URIs without slashes (e.g., 'timer:5s')
-            scheme, path = uri.split(':', 1)
-            uri = f"{scheme}://{path}"
-        
-        parsed = self._parse_uri(uri)
-        scheme = parsed['scheme']
-        
-        if scheme not in self.source_types:
-            raise ValueError(f"Unsupported source type: {scheme}")
-        
-        source_class = self.source_types[scheme]
-        try:
-            return source_class(uri, **parsed['options'])
-        except Exception as e:
-            raise ConnectorError(f"Failed to create source {scheme}: {e}") from e
+        """Create a source from a URI string."""
+        config = self._parse_uri_to_config(uri)
+        return self._create_source_from_config(config)
 
     def _create_destination_from_uri(self, uri: str) -> Destination:
-        """Create a destination connector from a URI.
-        
-        Args:
-            uri: Destination URI (e.g., 'http://example.com', 'file:/path/to/file')
-            
-        Returns:
-            Configured Destination instance
-        """
-        parsed = self._parse_uri(uri)
-        scheme = parsed['scheme']
+        """Create a destination from a URI string."""
+        config = self._parse_uri_to_config(uri)
+        return self._create_destination_from_config(config)
+    
+    def _parse_uri_to_config(self, uri: str) -> Dict[str, Any]:
+        """Parse a URI into a configuration dictionary."""
+        parsed = urlparse(uri)
+        scheme = parsed.scheme.lower()
 
-        if scheme not in self.destination_types:
-            raise ValueError(f"Unsupported destination type: {scheme}")
+        # Create base config
+        config = {
+            'uri': uri,
+            'scheme': scheme,
+            'path': parsed.path or ''
+        }
 
-        dest_class = self.destination_types[scheme]
-        try:
-            return dest_class(uri, **parsed['options'])
-        except Exception as e:
-            raise ConnectorError(f"Failed to create destination {scheme}: {e}") from e
+        # Add query parameters
+        if parsed.query:
+            query_params = parse_qs(parsed.query, keep_blank_values=True)
+            for key, value in query_params.items():
+                config[key] = value[0] if len(value) == 1 else value
+        
+        # Add optional components
+        if parsed.netloc:
+            config['netloc'] = parsed.netloc
+        if parsed.params:
+            config['params'] = dict(parse_qs(parsed.params))
+        if parsed.fragment:
+            config['fragment'] = parsed.fragment
+        if parsed.username:
+            config['username'] = parsed.username
+        if parsed.password:
+            config['password'] = parsed.password
+        if parsed.hostname:
+            config['hostname'] = parsed.hostname
+        if parsed.port is not None:
+            config['port'] = parsed.port
+            
+        return config
 
     def _create_source_from_config(self, config: Dict[str, Any]) -> Source:
-        """Create a source connector from a config dictionary.
-        
-        Args:
-            config: Source configuration dictionary
-            
-        Returns:
-            Configured Source instance
-        """
-        source_type = config.get('type')
-        if not source_type:
-            raise ValueError("Source config must include 'type'")
-        
-        if source_type not in self.source_types:
-            # Try to dynamically import the source
-            try:
-                module_name, _, class_name = source_type.rpartition('.')
-                if not module_name:
-                    raise ValueError(f"Invalid source type format: {source_type}")
-                
-                module = importlib.import_module(module_name)
-                source_class = getattr(module, class_name)
-                self.register_source(source_type, source_class)
-            except (ImportError, AttributeError) as e:
-                raise ValueError(f"Unknown source type: {source_type}") from e
-        
-        source_class = self.source_types[source_type]
+        """Create a source from a config dictionary."""
+        source_type = self._get_connector_type(config, 'source')
         try:
-            return source_class(**{k: v for k, v in config.items() if k != 'type'})
+            return self.source_types[source_type](config)
         except Exception as e:
-            raise ConnectorError(f"Failed to create source {source_type}: {e}") from e
+            raise ConnectorError(f"Failed to create source from config: {e}") from e
 
     def _create_destination_from_config(self, config: Dict[str, Any]) -> Destination:
-        """Create a destination connector from a config dictionary.
-        
-        Args:
-            config: Destination configuration dictionary
-            
-        Returns:
-            Configured Destination instance
-        """
-        dest_type = config.get('type')
-        if not dest_type:
-            raise ValueError("Destination config must include 'type'")
-        
-        if dest_type not in self.destination_types:
-            # Try to dynamically import the destination
-            try:
-                module_name, _, class_name = dest_type.rpartition('.')
-                if not module_name:
-                    raise ValueError(f"Invalid destination type format: {dest_type}")
-                
-                module = importlib.import_module(module_name)
-                dest_class = getattr(module, class_name)
-                self.register_destination(dest_type, dest_class)
-            except (ImportError, AttributeError) as e:
-                raise ValueError(f"Unknown destination type: {dest_type}") from e
-        
-        dest_class = self.destination_types[dest_type]
+        """Create a destination from a config dictionary."""
+        dest_type = self._get_connector_type(config, 'destination')
         try:
-            return dest_class(**{k: v for k, v in config.items() if k != 'type'})
+            return self.destination_types[dest_type](config)
         except Exception as e:
-            raise ConnectorError(f"Failed to create destination {dest_type}: {e}") from e
+            raise ConnectorError(f"Failed to create destination from config: {e}") from e
 
-    @staticmethod
-    def _parse_uri(uri: str) -> Dict[str, Any]:
-        """Parse a URI into its components and query parameters.
-        
-        Args:
-            uri: URI to parse
-            
-        Returns:
-            Dictionary with 'scheme', 'netloc', 'path', 'params', 'query', 'fragment',
-            and 'options' (query parameters as a dict)
-        """
-        if '://' not in uri and ':' in uri:
-            # Handle simple scheme:path format
-            scheme, path = uri.split(':', 1)
-            return {
-                'scheme': scheme,
-                'netloc': '',
-                'path': path,
-                'params': '',
-                'query': '',
-                'fragment': '',
-                'options': {}
-            }
-        
-        parsed = urlparse(uri)
-        options = {}
-        
-        # Parse query parameters
-        if parsed.query:
-            query_params = parse_qs(parsed.query, keep_blank_values=True)
-            options = {k: v[0] if len(v) == 1 else v for k, v in query_params.items()}
+    def _get_connector_type(self, config: Dict[str, Any], connector_kind: str) -> str:
+        """Get the connector type from config and validate it exists."""
+        if not isinstance(config, dict):
+            raise ValueError(f"{connector_kind.capitalize()} config must be a dictionary")
 
-        # Add username/password from netloc if present
-        if '@' in parsed.netloc:
-            auth_part, netloc = parsed.netloc.rsplit('@', 1)
-            if ':' in auth_part:
-                username, password = auth_part.split(':', 1)
-                options['username'] = username
-                options['password'] = password
+        connector_type = config.get('type') or config.get('scheme')
+        if not connector_type:
+            raise ValueError(f"{connector_kind.capitalize()} config must include 'type' or 'scheme'")
 
-        return {
-            'scheme': parsed.scheme,
-            'netloc': parsed.netloc,
-            'path': parsed.path,
-            'params': parsed.params,
-            'query': parsed.query,
-            'fragment': parsed.fragment,
-            'options': options
-        }
+        if connector_type not in getattr(self, f"{connector_kind}_types"):
+            raise ValueError(f"Unknown {connector_kind} type: {connector_type}")
+            
+        return connector_type
+    
+    def get_source_schemes(self) -> List[str]:
+        """Get a list of registered source schemes."""
+        return list(self.source_types.keys())
+    
+    def get_destination_schemes(self) -> List[str]:
+        """Get a list of registered destination schemes."""
+        return list(self.destination_types.keys())
+    
+    async def close(self) -> None:
+        """Close all connectors and release resources."""
+        # No resources to clean up in the base implementation
+        pass
 
 # Default connector manager instance
 default_connector_manager = ConnectorManager()

src/dialogchain/engine/route.py

@@ -317,47 +317,18 @@ async def _handle_error(self, error: Exception, message: Any = None) -> bool:
         return handled
 
     async def _safe_receive(self, source: Source) -> Any:
-        """Safely receive data from a source with retries."""
-        for attempt in range(self.retry_attempts + 1):
-            try:
-                return await asyncio.wait_for(
-                    source.receive(),
-                    timeout=self.timeout
-                )
-            except asyncio.TimeoutError:
-                if attempt == self.retry_attempts:
-                    logger.error(f"Timeout receiving from source in route {self.name}")
-                    raise
-                await asyncio.sleep(self.retry_delay)
-            except Exception as e:
-                logger.error(f"Error receiving from source in route {self.name}: {e}")
-                if attempt == self.retry_attempts:
-                    raise
-                await asyncio.sleep(self.retry_delay)
-        return None
-    
-    async def _safe_send(self, destination: Destination, data: Any) -> None:
-        """Safely send data to a destination with retries."""
-        for attempt in range(self.retry_attempts + 1):
-            try:
-                await asyncio.wait_for(
-                    destination.send(data),
-                    timeout=self.timeout
-                )
-                return
-            except asyncio.TimeoutError:
-                if attempt == self.retry_attempts:
-                    logger.error(f"Timeout sending to destination in route {self.name}")
-                    raise
-                await asyncio.sleep(self.retry_delay)
-            except Exception as e:
-                logger.error(f"Error sending to destination in route {self.name}: {e}")
-                if attempt == self.retry_attempts:
-                    raise
-                await asyncio.sleep(self.retry_delay)
-    
-    async def _safe_receive(self, source: Source) -> Any:
-        """Safely receive data from a source with retries."""
+        """Safely receive data from a source with retries.
+        
+        Args:
+            source: The source to receive data from
+            
+        Returns:
+            The received data, or None if all retries are exhausted
+            
+        Raises:
+            asyncio.TimeoutError: If the operation times out after all retries
+            Exception: If an error occurs after all retries
+        """
         for attempt in range(self.retry_attempts + 1):
             try:
                 return await asyncio.wait_for(
@@ -377,7 +348,16 @@ async def _safe_receive(self, source: Source) -> Any:
         return None
 
     async def _safe_send(self, destination: Destination, data: Any) -> None:
-        """Safely send data to a destination with retries."""
+        """Safely send data to a destination with retries.
+        
+        Args:
+            destination: The destination to send data to
+            data: The data to send
+            
+        Raises:
+            asyncio.TimeoutError: If the operation times out after all retries
+            Exception: If an error occurs after all retries
+        """
         for attempt in range(self.retry_attempts + 1):
             try:
                 await asyncio.wait_for(