diff --git a/src/cleveragents/reactive/config_parser.py b/src/cleveragents/reactive/config_parser.py index dd47fae5..fe5f8eb3 100644 --- a/src/cleveragents/reactive/config_parser.py +++ b/src/cleveragents/reactive/config_parser.py @@ -4,6 +4,7 @@ Configuration parser for RxPy-based CleverAgents. This module handles parsing of the new YAML configuration format that supports full RxPy reactive stream capabilities. """ +# pylint: disable=duplicate-code import logging import os @@ -11,18 +12,18 @@ import re from dataclasses import dataclass from dataclasses import field from pathlib import Path +from re import Match from typing import Any from typing import Dict from typing import List from typing import Optional -import yaml +import yaml from cleveragents.core.exceptions import ConfigurationError from cleveragents.reactive.route import BridgeConfig from cleveragents.reactive.route import RouteConfig from cleveragents.reactive.route import RouteType -from cleveragents.reactive.stream_router import StreamConfig from cleveragents.reactive.stream_router import StreamType from cleveragents.templates.yaml_template_engine import YAMLTemplateEngine @@ -37,7 +38,7 @@ class AgentConfig: @dataclass -class LangGraphConfig: +class LangGraphConfig: # pylint: disable=too-many-instance-attributes """Configuration for a LangGraph.""" name: str @@ -63,7 +64,7 @@ class HybridPipelineConfig: @dataclass -class ReactiveConfig: +class ReactiveConfig: # pylint: disable=too-many-instance-attributes """Complete reactive configuration.""" agents: Dict[str, AgentConfig] = field(default_factory=dict) @@ -82,7 +83,7 @@ class ReactiveConfig: prompts: Dict[str, Any] = field(default_factory=dict) -class ReactiveConfigParser: +class ReactiveConfigParser: # pylint: disable=too-few-public-methods """Parser for reactive configuration files.""" def __init__(self) -> None: @@ -91,13 +92,13 @@ class ReactiveConfigParser: def parse_files(self, config_files: List[Path]) -> ReactiveConfig: """Parse configuration from multiple files.""" combined_config: Dict[str, Any] = {} - all_template_sections: Dict[str, Any] = {} + _all_template_sections: Dict[str, Any] = {} # Reserved for future use for file_path in config_files: - self.logger.info(f"Loading configuration from {file_path}") + self.logger.info("Loading configuration from %s", file_path) # Check if this is a template file by looking for Jinja2 syntax - with open(file_path, "r") as f: + with open(file_path, "r", encoding="utf-8") as f: content = f.read() if "{%" in content or "{{" in content: @@ -117,7 +118,7 @@ class ReactiveConfigParser: return self._build_reactive_config(combined_config) - def _merge_configs(self, base: Dict[str, Any], new: Dict[str, Any]): + def _merge_configs(self, base: Dict[str, Any], new: Dict[str, Any]) -> None: """Merge configuration dictionaries.""" if new is None: return @@ -142,7 +143,7 @@ class ReactiveConfigParser: # Pattern to match ${VAR_NAME} or ${VAR_NAME:default_value} env_var_pattern = r"\${([A-Za-z0-9_]+)(?::([^}]*))?\}" - def replace_env_var(match): + def replace_env_var(match: Match[str]) -> str: env_var = match.group(1) default_value = match.group(2) if match.group(2) is not None else None @@ -152,16 +153,14 @@ class ReactiveConfigParser: # Convert default value to appropriate type if default_value.lower() in ("true", "false"): return str(default_value.lower() == "true") - elif default_value.isdigit(): + if default_value.isdigit(): return default_value - elif default_value.replace(".", "").isdigit(): + if default_value.replace(".", "").isdigit(): return default_value - else: - return default_value - else: - raise ConfigurationError( - f"Environment variable '{env_var}' is not set" - ) + return default_value + raise ConfigurationError( + f"Environment variable '{env_var}' is not set" + ) return env_value config = re.sub(env_var_pattern, replace_env_var, config) @@ -169,13 +168,15 @@ class ReactiveConfigParser: # Convert numeric and boolean strings to appropriate types if config.lower() in ("true", "false"): return config.lower() == "true" - elif config.isdigit(): + if config.isdigit(): return int(config) - elif config.replace(".", "").isdigit() and config.count(".") == 1: + if config.replace(".", "").isdigit() and config.count(".") == 1: return float(config) return config - def _build_reactive_config(self, config_dict: Dict[str, Any]) -> ReactiveConfig: + def _build_reactive_config( # pylint: disable=too-many-locals,too-many-branches + self, config_dict: Dict[str, Any] + ) -> ReactiveConfig: """Build reactive configuration from dictionary.""" reactive_config = ReactiveConfig() @@ -232,11 +233,11 @@ class ReactiveConfigParser: route_type_str = route_data["type"].lower() try: route_type = RouteType(route_type_str) - except ValueError: + except ValueError as exc: raise ConfigurationError( f"Route '{name}' has invalid type '{route_type_str}'. " f"Must be one of: stream, graph, bridge" - ) + ) from exc # Handle template instances if "template" in route_data or "route_template" in route_data: @@ -368,7 +369,9 @@ class ReactiveConfigParser: metadata=route_data.get("metadata", {}), ) - def _validate_config(self, config: ReactiveConfig): + def _validate_config( # pylint: disable=too-many-locals,too-many-branches + self, config: ReactiveConfig + ) -> None: """Validate the reactive configuration.""" # Collect all route names all_route_names = set(config.routes.keys()) | { @@ -404,7 +407,8 @@ class ReactiveConfigParser: for node_name, node_data in route_config.nodes.items(): if "agent" in node_data and node_data["agent"] not in config.agents: raise ConfigurationError( - f"Route '{route_name}' node '{node_name}' references unknown agent '{node_data['agent']}'" + f"Route '{route_name}' node '{node_name}' references " + f"unknown agent '{node_data['agent']}'" ) # Removed legacy stream validation - use routes instead @@ -459,7 +463,9 @@ class ReactiveConfigParser: if graph_name and not graph_found: # Graph might be defined inline, so only warn self.logger.warning( - f"Pipeline '{pipeline_name}' references graph '{graph_name}' not found in routes" + "Pipeline '%s' references graph '%s' not found in routes", + pipeline_name, + graph_name, ) elif stage_type == "stream": @@ -467,7 +473,9 @@ class ReactiveConfigParser: stream_name = stage.get("name", "") if stream_name and stream_name in config.routes: self.logger.warning( - f"Pipeline '{pipeline_name}' defines route '{stream_name}' that already exists" + "Pipeline '%s' defines route '%s' that already exists", + pipeline_name, + stream_name, ) self.logger.info("Configuration validation completed successfully") diff --git a/src/cleveragents/reactive/route.py b/src/cleveragents/reactive/route.py index ba727d8a..775cc922 100644 --- a/src/cleveragents/reactive/route.py +++ b/src/cleveragents/reactive/route.py @@ -5,8 +5,8 @@ This module provides a unified abstraction for both reactive streams (RxPy) and stateful graphs (LangGraph), treating them as different types of routes for data flow through processing pipelines. """ +# pylint: disable=duplicate-code -import logging from dataclasses import dataclass from dataclasses import field from enum import Enum @@ -14,7 +14,6 @@ from typing import Any from typing import Dict from typing import List from typing import Optional -from typing import Union from cleveragents.core.exceptions import ConfigurationError from cleveragents.langgraph.graph import GraphConfig @@ -52,7 +51,7 @@ class BridgeConfig: @dataclass -class RouteConfig: +class RouteConfig: # pylint: disable=too-many-instance-attributes """Unified configuration for all route types.""" name: str @@ -117,7 +116,7 @@ class RouteConfig: if self.type != RouteType.GRAPH: raise ValueError(f"Cannot convert {self.type} route to GraphConfig") - from pathlib import Path + from pathlib import Path # pylint: disable=import-outside-toplevel # Convert node dictionaries to NodeConfig objects node_configs = {} @@ -250,10 +249,9 @@ class RouteComplexityAnalyzer: """Analyze a route and return complexity metrics.""" if config.type == RouteType.STREAM: return RouteComplexityAnalyzer._analyze_stream(config) - elif config.type == RouteType.GRAPH: + if config.type == RouteType.GRAPH: return RouteComplexityAnalyzer._analyze_graph(config) - else: - return {"complexity": "bridge", "score": 0} + return {"complexity": "bridge", "score": 0} @staticmethod def _analyze_stream(config: RouteConfig) -> Dict[str, Any]: @@ -324,20 +322,18 @@ class RouteComplexityAnalyzer: """Get recommendation for stream complexity.""" if score <= 2: return "Good for simple transformations and filtering" - elif score <= 5: + if score <= 5: return "Suitable for multi-step processing pipelines" - else: - return "Consider using a graph if you need conditional logic or state" + return "Consider using a graph if you need conditional logic or state" @staticmethod def _get_graph_recommendation(score: int) -> str: """Get recommendation for graph complexity.""" if score <= 10: return "Good for workflows with conditional logic" - elif score <= 15: + if score <= 15: return "Suitable for complex stateful workflows" - else: - return "Advanced setup - ensure you need all features" + return "Advanced setup - ensure you need all features" @staticmethod def suggest_route_type(requirements: Dict[str, Any]) -> RouteType: @@ -350,10 +346,9 @@ class RouteComplexityAnalyzer: if needs_persistence or (needs_state and needs_conditionals): return RouteType.GRAPH - elif needs_conditionals and not is_continuous: + if needs_conditionals and not is_continuous: return RouteType.GRAPH - elif is_stateless and is_continuous: - return RouteType.STREAM - else: - # Default to stream for simpler cases + if is_stateless and is_continuous: return RouteType.STREAM + # Default to stream for simpler cases + return RouteType.STREAM diff --git a/src/cleveragents/reactive/route_bridge.py b/src/cleveragents/reactive/route_bridge.py index 0e2b09da..789f5a00 100644 --- a/src/cleveragents/reactive/route_bridge.py +++ b/src/cleveragents/reactive/route_bridge.py @@ -11,9 +11,7 @@ from typing import Any from typing import Dict from typing import Optional -import rx -from rx import operators as ops -from rx.scheduler.eventloop import AsyncIOScheduler +from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined] from cleveragents.agents.base import Agent from cleveragents.langgraph.graph import GraphConfig @@ -22,7 +20,6 @@ from cleveragents.langgraph.nodes import Edge from cleveragents.langgraph.nodes import NodeConfig from cleveragents.langgraph.nodes import NodeType from cleveragents.langgraph.state import GraphState -from cleveragents.reactive.route import BridgeConfig from cleveragents.reactive.route import RouteConfig from cleveragents.reactive.route import RouteType from cleveragents.reactive.stream_router import ReactiveStreamRouter @@ -78,7 +75,7 @@ class RouteBridge: if "complexity_threshold" in conditions: # Analyze route complexity - from cleveragents.reactive.route import RouteComplexityAnalyzer + from cleveragents.reactive.route import RouteComplexityAnalyzer # pylint: disable=import-outside-toplevel analysis = RouteComplexityAnalyzer.analyze_route(route_config) if analysis["score"] >= conditions["complexity_threshold"]: @@ -88,7 +85,8 @@ class RouteBridge: # Evaluate custom predicate function predicate = conditions["custom_predicate"] if callable(predicate): - return predicate(message, route_config) + result = predicate(message, route_config) + return bool(result) return False @@ -132,7 +130,7 @@ class RouteBridge: message: StreamMessage, ) -> LangGraph: """Convert a stream route to a graph route.""" - self.logger.info(f"Upgrading stream route '{route_config.name}' to graph") + self.logger.info("Upgrading stream route '%s' to graph", route_config.name) # Create graph config from stream graph_config = self._create_graph_from_stream(route_config) @@ -159,7 +157,7 @@ class RouteBridge: # Preserve subscriptions if configured if route_config.bridge and route_config.bridge.preserve_subscriptions: # Re-subscribe to the same sources - for sub in route_config.subscriptions: + for _ in route_config.subscriptions: # This would need implementation in the graph pass @@ -177,7 +175,7 @@ class RouteBridge: graph: LangGraph, ) -> StreamConfig: """Convert a graph route back to a stream route.""" - self.logger.info(f"Downgrading graph route '{route_config.name}' to stream") + self.logger.info("Downgrading graph route '%s' to stream", route_config.name) # Create stream config from graph stream_config = self._create_stream_from_graph(route_config, graph) @@ -187,14 +185,14 @@ class RouteBridge: flattener = route_config.bridge.state_flattener if callable(flattener): final_state = graph.state_manager.get_state() - flattened_data = flattener(final_state) + _flattened_data = flattener(final_state) # This could be injected into the stream somehow # Preserve checkpointing info if configured if route_config.bridge and route_config.bridge.preserve_checkpointing: # Save final checkpoint if checkpoint_dir is set if graph.state_manager.checkpoint_dir: - graph.state_manager._save_checkpoint() + graph.state_manager._save_checkpoint() # pylint: disable=protected-access self._active_conversions[route_config.name] = { "type": "stream", @@ -261,7 +259,7 @@ class RouteBridge: agents = [] # Traverse graph nodes in topological order - topological_levels = graph._topological_levels() + topological_levels = graph._topological_levels() # pylint: disable=protected-access sorted_nodes = [] for level in sorted(topological_levels.keys()): sorted_nodes.extend(sorted(topological_levels[level])) @@ -298,7 +296,7 @@ class RouteBridge: publications=route_config.publications, ) - def _get_message_count(self, route_name: str) -> int: + def _get_message_count(self, _route_name: str) -> int: """Get the number of messages processed by a route.""" # This would need to be tracked somewhere # For now, return a placeholder diff --git a/src/cleveragents/reactive/stream_router.py b/src/cleveragents/reactive/stream_router.py index 6dbc690b..13c649f6 100644 --- a/src/cleveragents/reactive/stream_router.py +++ b/src/cleveragents/reactive/stream_router.py @@ -5,6 +5,7 @@ This module provides reactive stream routing capabilities using RxPy, allowing for complex message flows with splitting, merging, filtering, and transformation operations. """ +# pylint: disable=duplicate-code import asyncio import logging @@ -16,15 +17,14 @@ from typing import Callable from typing import Dict from typing import List from typing import Optional -from typing import Union import rx from rx import operators as ops -from rx.core import Observable -from rx.core import Observer -from rx.scheduler.eventloop import AsyncIOScheduler -from rx.subject import BehaviorSubject -from rx.subject import Subject +from rx.core import Observable as ObservableType # type: ignore[attr-defined] +from rx.core import Observer as ObserverType # type: ignore[attr-defined] +from rx.scheduler.eventloop import AsyncIOScheduler # type: ignore[attr-defined] +from rx.subject import BehaviorSubject # type: ignore[attr-defined] +from rx.subject import Subject # type: ignore[attr-defined] from cleveragents.agents.base import Agent from cleveragents.core.exceptions import StreamRoutingError @@ -47,7 +47,7 @@ class StreamMessage: source_stream: Optional[str] = None timestamp: Optional[float] = None - def copy_with(self, **kwargs) -> "StreamMessage": + def copy_with(self, **kwargs: Any) -> "StreamMessage": """Create a copy with modified fields.""" data = { "content": self.content, @@ -60,7 +60,7 @@ class StreamMessage: @dataclass -class StreamConfig: +class StreamConfig: # pylint: disable=too-many-instance-attributes """Configuration for a reactive stream.""" name: str @@ -74,7 +74,7 @@ class StreamConfig: template_config: Optional[Dict[str, Any]] = None # Template configuration -class ReactiveStreamRouter: +class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes """ RxPy-based stream router for agent orchestration. @@ -99,7 +99,7 @@ class ReactiveStreamRouter: {} ) # Can be Subject, BehaviorSubject, or ReplaySubject self.stream_configs: Dict[str, StreamConfig] = {} - self.observables: Dict[str, Observable] = {} + self.observables: Dict[str, ObservableType] = {} # Agent registry self.agents: Dict[str, Agent] = {} @@ -139,10 +139,10 @@ class ReactiveStreamRouter: name="__error__", type=StreamType.COLD ) - def register_agent(self, name: str, agent: Agent): + def register_agent(self, name: str, agent: Agent) -> None: """Register an agent for use in streams.""" self.agents[name] = agent - self.logger.debug(f"Registered agent: {name}") + self.logger.debug("Registered agent: %s", name) def create_stream( self, config: StreamConfig @@ -156,7 +156,7 @@ class ReactiveStreamRouter: if config.type == StreamType.HOT: stream = BehaviorSubject(config.initial_value) elif config.type == StreamType.REPLAY: - from rx.subject import ReplaySubject + from rx.subject import ReplaySubject # type: ignore[attr-defined] # pylint: disable=import-outside-toplevel stream = ReplaySubject(buffer_size=config.buffer_size) else: # COLD @@ -172,15 +172,15 @@ class ReactiveStreamRouter: # Set up subscriptions self._setup_subscriptions(config) - self.logger.info(f"Created stream: {config.name} ({config.type.value})") + self.logger.info("Created stream: %s (%s)", config.name, config.type.value) return stream def _build_observable( self, stream_name: str, operator_configs: List[Dict[str, Any]] - ) -> Observable: + ) -> ObservableType: """Build an observable with configured operators.""" base_stream = self.streams[stream_name] - observable: Observable = base_stream + observable: ObservableType = base_stream for op_config in operator_configs: operator = self._create_operator(op_config) @@ -188,7 +188,7 @@ class ReactiveStreamRouter: return observable - def _create_operator(self, config: Dict[str, Any]): + def _create_operator(self, config: Dict[str, Any]) -> Any: # pylint: disable=too-many-return-statements,too-many-branches,too-many-statements,too-many-locals """Create an RxPy operator from configuration.""" op_type = config["type"] params = config.get("params", {}) @@ -220,43 +220,47 @@ class ReactiveStreamRouter: if not agent: raise StreamRoutingError(f"Agent '{agent_name}' not found") return ops.map(self._create_agent_mapper(agent)) - elif "function" in params: + if "function" in params: func_name = params["function"] return ops.map(getattr(self, f"_builtin_{func_name}", lambda x: x)) - elif "transform" in params: + if "transform" in params: transform = params["transform"] return ops.map(lambda x: self._apply_transform(x, transform)) + raise StreamRoutingError( + "Map operator requires 'agent', 'function', or 'transform' parameter" + ) # Filter operations - elif op_type == "filter": + if op_type == "filter": if "condition" in params: condition = params["condition"] return ops.filter(lambda x: self._evaluate_condition(x, condition)) + raise StreamRoutingError("Filter operator requires 'condition' parameter") # Timing operations - elif op_type == "debounce": + if op_type == "debounce": duration = params.get("duration", 1.0) # For single-shot mode, reduce debounce time debounce_time = min(duration, 0.05) # Max 50ms debounce for single-shot return ops.debounce(debounce_time) - elif op_type == "throttle": + if op_type == "throttle": duration = params.get("duration", 0.2) return ops.throttle_first(duration) - elif op_type == "delay": + if op_type == "delay": duration = params.get("duration", 0.1) return ops.delay(duration) # Buffering operations - modified for single-shot compatibility - elif op_type == "buffer": + if op_type == "buffer": if "count" in params: count = params["count"] timeout = params.get("timeout", 0.1) # For single-shot mode, use timeout to avoid hanging on insufficient messages # Create a custom operator that handles the buffer output properly - def buffer_and_flatten(source): + def buffer_and_flatten(source: ObservableType) -> ObservableType: return source.pipe( ops.buffer_with_time_or_count(timespan=timeout, count=count), ops.flat_map( @@ -265,10 +269,10 @@ class ReactiveStreamRouter: ) return buffer_and_flatten - elif "time" in params: + if "time" in params: timeout = params["time"] - def buffer_time_and_flatten(source): + def buffer_time_and_flatten(source: ObservableType) -> ObservableType: return source.pipe( ops.buffer_with_time(timeout), ops.flat_map( @@ -277,59 +281,60 @@ class ReactiveStreamRouter: ) return buffer_time_and_flatten + raise StreamRoutingError("Buffer operator requires 'count' or 'time' parameter") # Aggregation operations - elif op_type == "scan": + if op_type == "scan": if "accumulator" in params: return ops.scan( lambda acc, x: self._apply_accumulator( acc, x, params["accumulator"] ) ) + raise StreamRoutingError("Scan operator requires 'accumulator' parameter") - elif op_type == "reduce": + if op_type == "reduce": if "accumulator" in params: return ops.reduce( lambda acc, x: self._apply_accumulator( acc, x, params["accumulator"] ) ) + raise StreamRoutingError("Reduce operator requires 'accumulator' parameter") # Error handling - elif op_type == "catch": + if op_type == "catch": return ops.catch(self._handle_stream_error) - elif op_type == "retry": + if op_type == "retry": count = params.get("count", 3) return ops.retry(count) # Utility operations - elif op_type == "distinct": + if op_type == "distinct": return ops.distinct() - elif op_type == "take": + if op_type == "take": count = params.get("count", 1) return ops.take(count) - elif op_type == "skip": + if op_type == "skip": count = params.get("count", 1) return ops.skip(count) - elif op_type == "sample": + if op_type == "sample": interval = params.get("interval", 1.0) return ops.sample(interval) - else: - raise StreamRoutingError(f"Unknown operator type: {op_type}") + raise StreamRoutingError(f"Unknown operator type: {op_type}") - def _create_agent_mapper(self, agent: Agent) -> Callable: + def _create_agent_mapper(self, agent: Agent) -> Callable[[StreamMessage], StreamMessage]: """Create a mapper function for an agent.""" # For RxPY compatibility, we need to handle async operations def mapper(msg: StreamMessage) -> StreamMessage: - import asyncio - import concurrent.futures - import threading + import asyncio as async_io # pylint: disable=reimported,import-outside-toplevel + import threading # pylint: disable=import-outside-toplevel # Handle None message if msg is None: @@ -345,14 +350,14 @@ class ReactiveStreamRouter: def run_async() -> None: try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) + loop = async_io.new_event_loop() + async_io.set_event_loop(loop) try: result = loop.run_until_complete(agent.process_message(content)) result_holder.append(result) finally: loop.close() - except Exception as e: + except Exception as e: # pylint: disable=broad-exception-caught error_holder.append(e) # Run in thread with timeout @@ -361,10 +366,15 @@ class ReactiveStreamRouter: thread.join(timeout=30.0) # 30 second timeout if thread.is_alive(): - self.logger.error(f"Agent {agent.name} processing timed out") + self.logger.error("Agent %s processing timed out", agent.name) return msg.copy_with( content="Processing timed out", - metadata={**msg.metadata, "processed_by": agent.name, "error": True, "timeout": True} + metadata={ + **msg.metadata, + "processed_by": agent.name, + "error": True, + "timeout": True, + } ) if error_holder: @@ -375,8 +385,10 @@ class ReactiveStreamRouter: else: result = "No result returned" - except Exception as e: - self.logger.error(f"Failed to process message with {agent.name}: {e}") + except Exception as e: # pylint: disable=broad-exception-caught + self.logger.error( + "Failed to process message with %s: %s", agent.name, e + ) return msg.copy_with( content=f"Error: {str(e)}", metadata={**msg.metadata, "processed_by": agent.name, "error": True} @@ -391,21 +403,21 @@ class ReactiveStreamRouter: def _apply_transform( self, msg: StreamMessage, transform: Dict[str, Any] - ) -> StreamMessage: + ) -> StreamMessage: # pylint: disable=too-many-return-statements """Apply a transformation to a message.""" transform_type = transform.get("type", "identity") if transform_type == "extract": - field = transform.get("field") - if field and msg.content is not None and isinstance(msg.content, dict): - return msg.copy_with(content=msg.content.get(field)) + field_name = transform.get("field") + if field_name and msg.content is not None and isinstance(msg.content, dict): + return msg.copy_with(content=msg.content.get(field_name)) - elif transform_type == "wrap": + if transform_type == "wrap": wrapper = transform.get("wrapper", {}) content_data = msg.content if msg.content is not None else "" return msg.copy_with(content={**wrapper, "data": content_data}) - elif transform_type == "format": + if transform_type == "format": template = transform.get("template", "{content}") content_data = msg.content if msg.content is not None else "" formatted = template.format(content=content_data) @@ -413,7 +425,7 @@ class ReactiveStreamRouter: return msg - def _evaluate_condition( + def _evaluate_condition( # pylint: disable=too-many-return-statements self, msg: StreamMessage, condition: Dict[str, Any] ) -> bool: """Evaluate a filter condition.""" @@ -421,21 +433,21 @@ class ReactiveStreamRouter: if condition_type == "always": return True - elif condition_type == "never": + if condition_type == "never": return False - elif condition_type == "content_contains": + if condition_type == "content_contains": text = condition.get("text", "") content_str = str(msg.content) if msg.content is not None else "" return text in content_str - elif condition_type == "content_not_contains": + if condition_type == "content_not_contains": text = condition.get("text", "") content_str = str(msg.content) if msg.content is not None else "" return text not in content_str - elif condition_type == "metadata_has": + if condition_type == "metadata_has": key = condition.get("key", "") return key in msg.metadata - elif condition_type == "source_is": - source = condition.get("source", "") + if condition_type == "source_is": + source: str = condition.get("source", "") return msg.source_stream == source return True @@ -452,13 +464,13 @@ class ReactiveStreamRouter: acc.append(msg.content if msg.content is not None else "") return acc - elif acc_type == "concat": + if acc_type == "concat": if acc is None: acc = "" content_str = str(msg.content) if msg.content is not None else "" return acc + content_str - elif acc_type == "sum": + if acc_type == "sum": if acc is None: acc = 0 if msg.content is not None and isinstance(msg.content, (int, float)): @@ -467,7 +479,7 @@ class ReactiveStreamRouter: return acc - def _handle_stream_error(self, error: Exception, source: Observable) -> Observable: + def _handle_stream_error(self, error: Exception, source: ObservableType) -> ObservableType: """Handle stream errors.""" error_msg = StreamMessage( content=f"Stream error: {str(error)}", @@ -476,7 +488,7 @@ class ReactiveStreamRouter: self.streams["__error__"].on_next(error_msg) return source - def _setup_subscriptions(self, config: StreamConfig): + def _setup_subscriptions(self, config: StreamConfig) -> None: """Set up stream subscriptions.""" observable = self.observables[config.name] @@ -495,22 +507,22 @@ class ReactiveStreamRouter: ) self.subscriptions.append(subscription) self.logger.debug( - f"Stream '{config.name}' publishing to '{pub_stream_name}'" + "Stream '%s' publishing to '%s'", config.name, pub_stream_name ) - def merge_streams(self, stream_names: List[str], output_stream_name: str): + def merge_streams(self, stream_names: List[str], output_stream_name: str) -> None: """Merge multiple streams into one.""" # Check if output stream already exists if output_stream_name in self.streams: # If it exists, we'll merge into the existing stream output_stream = self.streams[output_stream_name] - self.logger.debug(f"Merging into existing stream: {output_stream_name}") + self.logger.debug("Merging into existing stream: %s", output_stream_name) else: # Create new output stream if it doesn't exist output_stream = Subject() self.streams[output_stream_name] = output_stream self.observables[output_stream_name] = output_stream - self.logger.debug(f"Created new stream for merge: {output_stream_name}") + self.logger.debug("Created new stream for merge: %s", output_stream_name) source_observables = [] for name in stream_names: @@ -529,11 +541,11 @@ class ReactiveStreamRouter: subscription = merged_observable.subscribe(output_stream) self.subscriptions.append(subscription) - self.logger.info(f"Merged streams {stream_names} into {output_stream_name}") + self.logger.info("Merged streams %s into %s", stream_names, output_stream_name) def split_stream( self, source_stream_name: str, conditions: Dict[str, Dict[str, Any]] - ): + ) -> None: """Split a stream into multiple streams based on conditions.""" if source_stream_name not in self.observables: raise StreamRoutingError(f"Source stream '{source_stream_name}' not found") @@ -550,8 +562,11 @@ class ReactiveStreamRouter: output_stream = Subject() self.streams[output_name] = output_stream - # Create filtered observable - filter_op = ops.filter(lambda x: self._evaluate_condition(x, condition)) + # Create filtered observable with closure to avoid cell-var-from-loop + def make_filter(cond: Dict[str, Any]) -> Any: + return ops.filter(lambda x: self._evaluate_condition(x, cond)) + + filter_op = make_filter(condition) filtered_observable = source_observable.pipe(filter_op) self.observables[output_name] = filtered_observable @@ -560,12 +575,12 @@ class ReactiveStreamRouter: self.subscriptions.append(subscription) self.logger.info( - f"Split stream {source_stream_name} into {list(conditions.keys())}" + "Split stream %s into %s", source_stream_name, list(conditions.keys()) ) def send_message( self, stream_name: str, content: Any, metadata: Optional[Dict[str, Any]] = None - ): + ) -> None: """Send a message to a stream.""" if stream_name not in self.streams: raise StreamRoutingError(f"Stream '{stream_name}' not found") @@ -579,11 +594,11 @@ class ReactiveStreamRouter: self.streams[stream_name].on_next(message) - def subscribe_to_output(self, observer: Observer): + def subscribe_to_output(self, observer: ObserverType) -> None: """Subscribe to the output stream.""" self.observables["__output__"].subscribe(observer) - def dispose(self): + def dispose(self) -> None: """Dispose of all subscriptions and streams.""" for subscription in self.subscriptions: subscription.dispose()