import json from pydantic_settings import BaseSettings from typing import List, Any, Dict, Optional, Union from pydantic import Field, field_validator, ValidationError class Settings(BaseSettings): # App Settings APP_NAME: str = Field(default="Search Coordinator Service", description="Name of the application", env="APP_NAME") VERSION: str = Field(default="1.0.0", description="Version of the application", env="VERSION") DEBUG: bool = Field(default=False, description="Enable debug mode", env="DEBUG") HOST: str = Field(default="0.0.0.0", description="Host for the application", env="HOST") PORT: int = Field(default=8000, description="Port for the application", env="PORT") ENVIRONMENT: str = Field(default="development", description="Environment type (development/production)", env="ENVIRONMENT") # Security Settings SECRET_KEY: str = Field(default="your-secret-key-here", description="Secret key for security", env="SECRET_KEY") ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(default=30, description="Access token expiration time in minutes", env="ACCESS_TOKEN_EXPIRE_MINUTES") ALGORITHM: str = Field(default="HS256", description="Algorithm used for token encoding", env="ALGORITHM") # CORS settings CORS_ORIGINS: List[str] = Field(default=["http://localhost:3000", "http://localhost:8080"], description="List of allowed origins for CORS", env="CORS_ORIGINS") CORS_ALLOW_CREDENTIALS: bool = Field(default=True, description="Allow credentials for CORS", env="CORS_ALLOW_CREDENTIALS") CORS_ALLOW_METHODS: List[str] = Field(default=["*"], description="List of allowed HTTP methods for CORS", env="CORS_ALLOW_METHODS") CORS_ALLOW_HEADERS: List[str] = Field(default=["*"], description="List of allowed HTTP headers for CORS", env="CORS_ALLOW_HEADERS") ALLOWED_HOSTS: List[str] = Field(default=["localhost", "127.0.0.1"], description="Allowed hosts for CORS", env="ALLOWED_HOSTS") # MongoDB Settings MONGODB_URL: str = Field(default="mongodb://mongodb:27017", description="MongoDB connection URL", env="MONGODB_URL") DATABASE_NAME: str = Field(default="search_coordinator", description="Database name for MongoDB", env="DATABASE_NAME") MONGODB_MIN_POOL_SIZE: int = Field(default=10, description="Minimum connection pool size for MongoDB", env="MONGODB_MIN_POOL_SIZE") MONGODB_MAX_POOL_SIZE: int = Field(default=100, description="Maximum connection pool size for MongoDB", env="MONGODB_MAX_POOL_SIZE") # Kafka Settings - Using Union to handle string or list input KAFKA_BOOTSTRAP_SERVERS: Union[str, List[str]] = Field( default="kafka:9092", description="Kafka bootstrap servers (comma-separated string or JSON list)", env="KAFKA_BOOTSTRAP_SERVERS" ) KAFKA_CONSUMER_GROUP: str = Field(default="search-coordinator-group", description="Kafka consumer group name", env="KAFKA_CONSUMER_GROUP") KAFKA_PRODUCER_BATCH_SIZE: int = Field(default=16384, description="Batch size for Kafka producer", env="KAFKA_PRODUCER_BATCH_SIZE") KAFKA_PRODUCER_LINGER_MS: int = Field(default=10, description="Linger time for Kafka producer in milliseconds", env="KAFKA_PRODUCER_LINGER_MS") KAFKA_CONSUMER_AUTO_OFFSET_RESET: str = Field(default="latest", description="Auto offset reset policy for Kafka consumer", env="KAFKA_CONSUMER_AUTO_OFFSET_RESET") # Queue Topics GOOGLE_QUEUE_TOPIC: str = Field(default="google-search-queue", description="Queue topic for Google searches", env="GOOGLE_QUEUE_TOPIC") YOUTUBE_QUEUE_TOPIC: str = Field(default="youtube-search-queue", description="Queue topic for YouTube searches", env="YOUTUBE_QUEUE_TOPIC") RSS_QUEUE_TOPIC: str = Field(default="rss-search-queue", description="Queue topic for RSS feeds", env="RSS_QUEUE_TOPIC") BING_QUEUE_TOPIC: str = Field(default="bing-search-queue", description="Queue topic for Bing searches", env="BING_QUEUE_TOPIC") DUCKDUCKGO_QUEUE_TOPIC: str = Field(default="duckduckgo-search-queue", description="Queue topic for DuckDuckGo searches", env="DUCKDUCKGO_QUEUE_TOPIC") REDDIT_QUEUE_TOPIC: str = Field(default="reddit-search-queue", description="Queue topic for Reddit searches", env="REDDIT_QUEUE_TOPIC") SOCIAL_MEDIA_QUEUE_TOPIC: str = Field(default="social-media-search-queue", description="Queue topic for social media searches", env="SOCIAL_MEDIA_QUEUE_TOPIC") SEARCH_RESULTS_TOPIC: str = Field(default="search-results", description="Queue topic for search results", env="SEARCH_RESULTS_TOPIC") # Redis Settings REDIS_URL: str = Field(default="redis://redis:6379", description="Redis connection URL", env="REDIS_URL") REDIS_MAX_CONNECTIONS: int = Field(default=20, description="Maximum connections to Redis", env="REDIS_MAX_CONNECTIONS") REDIS_RETRY_ON_TIMEOUT: bool = Field(default=True, description="Retry on timeout for Redis connections", env="REDIS_RETRY_ON_TIMEOUT") # Scheduler Settings PERIODIC_SEARCH_INTERVAL: int = Field(default=300, description="Interval for periodic searches in seconds", env="PERIODIC_SEARCH_INTERVAL") MAX_CONCURRENT_SEARCHES: int = Field(default=100, description="Maximum concurrent searches allowed", env="MAX_CONCURRENT_SEARCHES") SEARCH_TIMEOUT: int = Field(default=300, description="Timeout for search requests in seconds", env="SEARCH_TIMEOUT") # Rate Limiting RATE_LIMIT_REQUESTS: int = Field(default=100, description="Maximum number of requests allowed", env="RATE_LIMIT_REQUESTS") RATE_LIMIT_WINDOW: int = Field(default=60, description="Time window for rate limiting in seconds", env="RATE_LIMIT_WINDOW") # Monitoring METRICS_PORT: int = Field(default=8001, description="Port for metrics exposure", env="METRICS_PORT") ENABLE_METRICS: bool = Field(default=True, description="Enable metrics collection", env="ENABLE_METRICS") LOG_LEVEL: str = Field(default="INFO", description="Logging level for the application", env="LOG_LEVEL") # External Services NOTIFICATION_SERVICE_URL: Optional[str] = Field(default=None, description="URL for notification service", env="NOTIFICATION_SERVICE_URL") USER_SERVICE_URL: Optional[str] = Field(default=None, description="URL for user service", env="USER_SERVICE_URL") # Performance Settings MAX_WORKERS: int = Field(default=4, description="Maximum number of workers", env="MAX_WORKERS") WORKER_TIMEOUT: int = Field(default=30, description="Timeout for worker processes in seconds", env="WORKER_TIMEOUT") # Feature Flags ENABLE_CACHING: bool = Field(default=True, description="Enable caching for the application", env="ENABLE_CACHING") ENABLE_RATE_LIMITING: bool = Field(default=True, description="Enable rate limiting", env="ENABLE_RATE_LIMITING") ENABLE_METRICS_COLLECTION: bool = Field(default=True, description="Enable metrics collection", env="ENABLE_METRICS_COLLECTION") ENABLE_PERIODIC_CLEANUP: bool = Field(default=True, description="Enable periodic cleanup", env="ENABLE_PERIODIC_CLEANUP") # Cleanup Settings CLEANUP_INTERVAL: int = Field(default=3600, description="Interval for cleanup in seconds", env="CLEANUP_INTERVAL") COMPLETED_REQUESTS_RETENTION_DAYS: int = Field(default=30, description="Retention period for completed requests in days", env="COMPLETED_REQUESTS_RETENTION_DAYS") FAILED_REQUESTS_RETENTION_DAYS: int = Field(default=7, description="Retention period for failed requests in days", env="FAILED_REQUESTS_RETENTION_DAYS") class Config: env_file = ".env" case_sensitive = True env_file_encoding = 'utf-8' @field_validator('KAFKA_BOOTSTRAP_SERVERS') @classmethod def validate_kafka_bootstrap_servers(cls, value): """Validate and normalize Kafka bootstrap servers.""" # Handle None or empty values if value is None or value == "": return ["kafka:9092"] # If it's already a list, return it if isinstance(value, list): return [server.strip() for server in value if server.strip()] or ["kafka:9092"] # If it's a string, process it if isinstance(value, str): value = value.strip() # Handle empty string after stripping if not value: return ["kafka:9092"] # Handle comma-separated values if "," in value: servers = [server.strip() for server in value.split(",") if server.strip()] return servers if servers else ["kafka:9092"] # Handle JSON string if value.startswith('[') and value.endswith(']'): try: parsed = json.loads(value) if isinstance(parsed, list): return [str(server).strip() for server in parsed if str(server).strip()] or ["kafka:9092"] except json.JSONDecodeError: pass # Single server return [value] # Fallback to default return ["kafka:9092"] @property def kafka_bootstrap_servers_list(self) -> List[str]: """Get Kafka bootstrap servers as a list.""" if isinstance(self.KAFKA_BOOTSTRAP_SERVERS, list): return self.KAFKA_BOOTSTRAP_SERVERS return [self.KAFKA_BOOTSTRAP_SERVERS] @property def mongodb_url_with_options(self) -> str: """Get MongoDB URL with connection options.""" return f"{self.MONGODB_URL}?minPoolSize={self.MONGODB_MIN_POOL_SIZE}&maxPoolSize={self.MONGODB_MAX_POOL_SIZE}" @property def kafka_config(self) -> Dict[str, Any]: """Get Kafka configuration.""" return { "bootstrap_servers": self.kafka_bootstrap_servers_list, "batch_size": self.KAFKA_PRODUCER_BATCH_SIZE, "linger_ms": self.KAFKA_PRODUCER_LINGER_MS, "auto_offset_reset": self.KAFKA_CONSUMER_AUTO_OFFSET_RESET } # Instantiate settings with better error handling try: settings = Settings() print(f"✅ Settings loaded successfully!") print(f"📡 Kafka Bootstrap Servers: {settings.kafka_bootstrap_servers_list}") print(f"🔧 Environment: {settings.ENVIRONMENT}") print(f"🗄️ Database: {settings.DATABASE_NAME}") except ValidationError as e: print("❌ Validation Error loading settings:") for error in e.errors(): field = '.'.join(str(loc) for loc in error['loc']) print(f" - {field}: {error['msg']}") if 'input' in error: print(f" Input value: {repr(error['input'])}") raise except Exception as e: print(f"❌ Unexpected error loading settings: {e}") raise