""" Configuration management for Noveum Trace SDK. This module handles configuration loading from multiple sources including environment variables, configuration files, and programmatic configuration. """ import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional, Union import yaml from noveum_trace.utils.exceptions import ConfigurationError def _parse_config_bool(value: Any, *, field_name: str) -> bool: """ Coerce config dict values to bool (YAML/JSON often yield strings). Accepts bool, int (0/1), and common string forms; raises ConfigurationError for unrecognized values. """ if isinstance(value, bool): return value if value is None: return False if isinstance(value, int): if value == 0: return False if value == 1: return True raise ConfigurationError( f"Invalid boolean for {field_name}: {value!r} " "(expected 0 or 1 when using an integer)" ) if isinstance(value, str): s = value.strip().lower() if s in ("true", "1", "yes", "on"): return True if s in ("false", "0", "no", "off", ""): return False raise ConfigurationError( f"Invalid boolean for {field_name}: {value!r} " '(expected "true"/"false", "1"/"0", "yes"/"no", etc.)' ) raise ConfigurationError( f"Invalid boolean for {field_name}: {value!r} " f"(unsupported type {type(value).__name__})" ) # Configuration constants DEFAULT_ENDPOINT = "https://api.noveum.ai/api" DEFAULT_TIMEOUT = 30 DEFAULT_RETRY_ATTEMPTS = 3 DEFAULT_BATCH_SIZE = 100 DEFAULT_BATCH_TIMEOUT = 5.0 DEFAULT_MAX_QUEUE_SIZE = 1000 DEFAULT_MAX_SPANS_PER_TRACE = 1000 DEFAULT_DEV_TRACES_DIR = ".noveum_trace_dev" # Not a cryptographic default: absence of salt. Pseudonymization requires an # explicit deployment secret when ``SecurityConfig.pii_enabled`` is True. DEFAULT_PII_SALT: Optional[str] = None _DEPRECATED_SHARED_PII_SALT = "noveum-pii-default-salt" @dataclass class TracingConfig: """Configuration for tracing behavior.""" enabled: bool = True sample_rate: float = 1.0 max_spans_per_trace: int = DEFAULT_MAX_SPANS_PER_TRACE capture_errors: bool = True capture_stack_traces: bool = False capture_performance: bool = False @dataclass class TransportConfig: """Configuration for transport layer.""" endpoint: str = DEFAULT_ENDPOINT timeout: int = DEFAULT_TIMEOUT retry_attempts: int = DEFAULT_RETRY_ATTEMPTS retry_backoff: float = 1.0 batch_size: int = DEFAULT_BATCH_SIZE batch_timeout: float = DEFAULT_BATCH_TIMEOUT max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE compression: bool = True ssl_verify: bool = ( # Set to False to disable SSL verification (NOT recommended for production) True ) # Path to custom CA bundle for corporate proxies ca_bundle: Optional[str] = None @dataclass class SecurityConfig: """Configuration for security and privacy.""" redact_pii: bool = False custom_redaction_patterns: list[str] = field(default_factory=list) encrypt_data: bool = True data_residency: Optional[str] = None pii_enabled: bool = False pii_salt: Optional[str] = DEFAULT_PII_SALT @dataclass class IntegrationConfig: """Configuration for framework integrations.""" langchain: dict[str, Any] = field(default_factory=lambda: {"enabled": False}) llamaindex: dict[str, Any] = field(default_factory=lambda: {"enabled": False}) openai: dict[str, Any] = field(default_factory=lambda: {"enabled": False}) anthropic: dict[str, Any] = field(default_factory=lambda: {"enabled": False}) @dataclass class Config: """Main configuration class for Noveum Trace SDK.""" # Core settings project: Optional[str] = None api_key: Optional[str] = None environment: str = "development" service_version: Optional[str] = None # Component configurations tracing: TracingConfig = field(default_factory=TracingConfig) transport: TransportConfig = field(default_factory=TransportConfig) security: SecurityConfig = field(default_factory=SecurityConfig) integrations: IntegrationConfig = field(default_factory=IntegrationConfig) # Additional settings debug: bool = False log_level: str = "ERROR" dev_mode: bool = False dev_traces_dir: Optional[str] = None # Additively emit OpenTelemetry-aligned fields (gen_ai.* attrs, span kind, status # object, resource, scope, W3C-format ids) alongside the legacy payload. On by # default; set NOVEUM_OTEL_COMPAT=false to emit the exact legacy payload. otel_compat: bool = True # Private field to store endpoint override _endpoint_override: Optional[str] = field(default=None, init=False, repr=False) def __post_init__(self) -> None: """Initialize configuration after dataclass initialization.""" self._validate() @property def endpoint(self) -> str: """Get the current endpoint from transport configuration.""" return self.transport.endpoint @endpoint.setter def endpoint(self, value: str) -> None: """Set the endpoint in transport configuration.""" self.transport.endpoint = value @classmethod def create( cls, project: Optional[str] = None, api_key: Optional[str] = None, environment: str = "development", service_version: Optional[str] = None, endpoint: Optional[str] = None, tracing: Optional[TracingConfig] = None, transport: Optional[TransportConfig] = None, security: Optional[SecurityConfig] = None, integrations: Optional[IntegrationConfig] = None, debug: bool = False, log_level: str = "ERROR", dev_mode: bool = False, dev_traces_dir: Optional[str] = None, otel_compat: bool = True, ) -> "Config": """Create a Config instance with optional endpoint override.""" # Create the config instance config = cls( project=project, api_key=api_key, environment=environment, service_version=service_version, tracing=tracing or TracingConfig(), transport=transport or TransportConfig(), security=security or SecurityConfig(), integrations=integrations or IntegrationConfig(), debug=debug, log_level=log_level, dev_mode=dev_mode, dev_traces_dir=dev_traces_dir, otel_compat=otel_compat, ) # Handle endpoint override after initialization if endpoint is not None: config.transport.endpoint = endpoint return config def _validate(self) -> None: """Validate configuration values.""" if self.tracing.sample_rate < 0 or self.tracing.sample_rate > 1: raise ConfigurationError( f"Invalid sample_rate: {self.tracing.sample_rate}. " "Must be between 0 and 1." ) if self.tracing.max_spans_per_trace <= 0: raise ConfigurationError( f"Invalid max_spans_per_trace: {self.tracing.max_spans_per_trace}. " "Must be greater than 0." ) if self.transport.timeout <= 0: raise ConfigurationError( f"Invalid timeout: {self.transport.timeout}. Must be greater than 0." ) # Validate endpoint URL format endpoint = self.transport.endpoint if endpoint: # Check if it's a valid URL with proper scheme if not endpoint.startswith(("http://", "https://")): raise ConfigurationError( f"Invalid endpoint URL: {endpoint}. " "Must start with 'http://' or 'https://'" ) # Basic URL validation - check for invalid characters or patterns import re # Simple regex to validate basic URL structure url_pattern = r"^https?://[a-zA-Z0-9\-._~:/?#[\]@!$&\'()*+,;=%]+$" if not re.match(url_pattern, endpoint): raise ConfigurationError(f"Invalid endpoint URL format: {endpoint}") if self.security.pii_enabled: salt = self.security.pii_salt if salt is None: raise ConfigurationError( "security.pii_enabled is True but security.pii_salt is missing. " "Set a deployment-specific string secret (e.g. NOVEUM_PII_SALT or " "security.pii_salt in configuration)." ) if not isinstance(salt, str): raise ConfigurationError( "security.pii_enabled is True but security.pii_salt must be a str, " f"not {type(salt).__name__}. " "Use a string secret (e.g. quote the value in YAML/JSON)." ) stripped = salt.strip() if not stripped: raise ConfigurationError( "security.pii_enabled is True but security.pii_salt is empty or " "whitespace-only. Set a non-empty string secret." ) if stripped == _DEPRECATED_SHARED_PII_SALT: raise ConfigurationError( "security.pii_salt must not use the deprecated shared default value. " "Set a unique secret for your deployment." ) def to_dict(self) -> dict[str, Any]: """Convert configuration to dictionary.""" return { "project": self.project, "api_key": self.api_key, "environment": self.environment, "service_version": self.service_version, "tracing": { "enabled": self.tracing.enabled, "sample_rate": self.tracing.sample_rate, "max_spans_per_trace": self.tracing.max_spans_per_trace, "capture_errors": self.tracing.capture_errors, "capture_stack_traces": self.tracing.capture_stack_traces, "capture_performance": self.tracing.capture_performance, }, "transport": { "endpoint": self.transport.endpoint, "timeout": self.transport.timeout, "retry_attempts": self.transport.retry_attempts, "retry_backoff": self.transport.retry_backoff, "batch_size": self.transport.batch_size, "batch_timeout": self.transport.batch_timeout, "compression": self.transport.compression, "ssl_verify": self.transport.ssl_verify, "ca_bundle": self.transport.ca_bundle, }, "security": { "redact_pii": self.security.redact_pii, "custom_redaction_patterns": self.security.custom_redaction_patterns, "encrypt_data": self.security.encrypt_data, "data_residency": self.security.data_residency, "pii_enabled": self.security.pii_enabled, "pii_salt": self.security.pii_salt, }, "integrations": { "langchain": self.integrations.langchain, "llamaindex": self.integrations.llamaindex, "openai": self.integrations.openai, "anthropic": self.integrations.anthropic, }, "debug": self.debug, "log_level": self.log_level, "dev_mode": self.dev_mode, "dev_traces_dir": self.dev_traces_dir, "otel_compat": self.otel_compat, } @classmethod def from_dict(cls, data: dict[str, Any]) -> "Config": """Create configuration from dictionary.""" config = cls() # Core settings config.project = data.get("project") config.api_key = data.get("api_key") config.environment = data.get("environment", "development") config.service_version = data.get("service_version") config.debug = data.get("debug", False) config.log_level = data.get("log_level", "ERROR") if "otel_compat" in data: config.otel_compat = _parse_config_bool( data["otel_compat"], field_name="otel_compat" ) if "dev_mode" in data: config.dev_mode = _parse_config_bool( data["dev_mode"], field_name="dev_mode" ) else: config.dev_mode = False dev_traces_dir_raw = data.get("dev_traces_dir") if dev_traces_dir_raw is None: config.dev_traces_dir = None elif isinstance(dev_traces_dir_raw, str) and not dev_traces_dir_raw.strip(): config.dev_traces_dir = None else: config.dev_traces_dir = dev_traces_dir_raw # Handle top-level endpoint parameter top_level_endpoint = data.get("endpoint") # Tracing configuration if "tracing" in data: tracing_data = data["tracing"] if isinstance(tracing_data, dict): config.tracing = TracingConfig( enabled=tracing_data.get("enabled", True), sample_rate=tracing_data.get("sample_rate", 1.0), max_spans_per_trace=tracing_data.get("max_spans_per_trace", 1000), capture_errors=tracing_data.get("capture_errors", True), capture_stack_traces=tracing_data.get( "capture_stack_traces", False ), capture_performance=tracing_data.get("capture_performance", False), ) else: # If tracing is not a dict, log a warning and use default config.tracing = TracingConfig() # Transport configuration if "transport" in data: transport_data = data["transport"] if isinstance(transport_data, dict): config.transport = TransportConfig( endpoint=transport_data.get("endpoint", DEFAULT_ENDPOINT), timeout=transport_data.get("timeout", DEFAULT_TIMEOUT), retry_attempts=transport_data.get( "retry_attempts", DEFAULT_RETRY_ATTEMPTS ), retry_backoff=transport_data.get("retry_backoff", 1.0), batch_size=transport_data.get("batch_size", DEFAULT_BATCH_SIZE), batch_timeout=transport_data.get( "batch_timeout", DEFAULT_BATCH_TIMEOUT ), max_queue_size=transport_data.get( "max_queue_size", DEFAULT_MAX_QUEUE_SIZE ), compression=transport_data.get("compression", False), ssl_verify=transport_data.get("ssl_verify", True), ca_bundle=transport_data.get("ca_bundle"), ) else: # If transport is not a dict, use default config.transport = TransportConfig() # Handle top-level endpoint override if top_level_endpoint is not None: config.transport.endpoint = top_level_endpoint # Security configuration if "security" in data: security_data = data["security"] if isinstance(security_data, dict): config.security = SecurityConfig( redact_pii=security_data.get("redact_pii", False), custom_redaction_patterns=security_data.get( "custom_redaction_patterns", [] ), encrypt_data=security_data.get("encrypt_data", True), data_residency=security_data.get("data_residency"), pii_enabled=_parse_config_bool( security_data.get("pii_enabled", False), field_name="security.pii_enabled", ), pii_salt=security_data.get("pii_salt", DEFAULT_PII_SALT), ) else: # If security is not a dict, use default config.security = SecurityConfig() # Integration configuration if "integrations" in data: integrations_data = data["integrations"] if isinstance(integrations_data, dict): config.integrations = IntegrationConfig( langchain=integrations_data.get("langchain", {"enabled": False}), llamaindex=integrations_data.get("llamaindex", {"enabled": False}), openai=integrations_data.get("openai", {"enabled": False}), anthropic=integrations_data.get("anthropic", {"enabled": False}), ) else: # If integrations is not a dict, use default config.integrations = IntegrationConfig() # Validate the final configuration config._validate() return config # Global configuration instance _config: Optional[Config] = None def _deep_merge_dicts(base: dict[str, Any], update: dict[str, Any]) -> dict[str, Any]: """ Deep merge two dictionaries, with update values taking precedence. None values in update are ignored (don't override existing values). Args: base: The base dictionary to merge into update: The dictionary whose values should take precedence Returns: A new dictionary with merged values """ merged = base.copy() for key, value in update.items(): if value is None: # Skip None values - they should not override existing values continue elif ( key in merged and isinstance(merged[key], dict) and isinstance(value, dict) ): # Recursively merge nested dictionaries merged[key] = _deep_merge_dicts(merged[key], value) else: # For non-dict values or new keys, use the update value merged[key] = value return merged def configure( config_data: Optional[Union[dict[str, Any], Config, str, Path]] = None, ) -> None: """ Configure the Noveum Trace SDK. Args: config_data: Configuration data as dict, Config instance, file path, or None to load from environment """ global _config if config_data is None or (isinstance(config_data, dict) and not config_data): # Load from environment and default config files when None or empty dict _config = _load_from_environment() elif isinstance(config_data, Config): _config = config_data elif isinstance(config_data, dict): # Merge provided config with environment config # Explicit parameters take precedence over environment variables env_config = _load_from_environment() merged_data = env_config.to_dict() # Deep merge the config data with environment config # This preserves nested settings while allowing overrides merged_data = _deep_merge_dicts(merged_data, config_data) # Handle top-level endpoint parameter - it should override transport.endpoint # regardless of merge order for consistent behavior if "endpoint" in config_data and config_data["endpoint"] is not None: # Top-level endpoint overrides transport.endpoint if "transport" not in merged_data: merged_data["transport"] = {} merged_data["transport"]["endpoint"] = config_data["endpoint"] _config = Config.from_dict(merged_data) elif isinstance(config_data, (str, Path)): _config = _load_from_file(config_data) else: raise ConfigurationError( f"Invalid configuration type: {type(config_data)}. " "Expected dict, Config, str, or Path." ) def get_config() -> Config: """ Get the current configuration. Returns: Current Config instance Raises: ConfigurationError: If configuration has not been set """ global _config if _config is None: # Auto-configure from environment configure() assert _config is not None, "Configuration must be set after configure()" return _config def _load_from_environment() -> Config: """Load configuration from environment variables and config files.""" config_data: dict[str, Any] = {} # Load from environment variables if os.getenv("NOVEUM_PROJECT"): config_data["project"] = os.getenv("NOVEUM_PROJECT") if os.getenv("NOVEUM_API_KEY"): config_data["api_key"] = os.getenv("NOVEUM_API_KEY") if os.getenv("NOVEUM_ENVIRONMENT"): config_data["environment"] = os.getenv("NOVEUM_ENVIRONMENT") if os.getenv("NOVEUM_ENDPOINT"): # Set as top-level endpoint for consistency with programmatic API config_data["endpoint"] = os.getenv("NOVEUM_ENDPOINT") # SSL configuration from environment ssl_verify_env = os.getenv("NOVEUM_SSL_VERIFY") if ssl_verify_env is not None: if "transport" not in config_data: config_data["transport"] = {} config_data["transport"]["ssl_verify"] = ssl_verify_env.lower() not in ( "false", "0", "no", "off", ) ca_bundle_env = os.getenv("NOVEUM_CA_BUNDLE") if ca_bundle_env: if "transport" not in config_data: config_data["transport"] = {} config_data["transport"]["ca_bundle"] = ca_bundle_env if os.getenv("NOVEUM_SERVICE_VERSION"): config_data["service_version"] = os.getenv("NOVEUM_SERVICE_VERSION") debug_env = os.getenv("NOVEUM_DEBUG") if debug_env: config_data["debug"] = debug_env.lower() in ("true", "1", "yes") dev_mode_env = os.getenv("NOVEUM_DEV_MODE") if dev_mode_env: config_data["dev_mode"] = dev_mode_env.lower() in ("true", "1", "yes") otel_compat_env = os.getenv("NOVEUM_OTEL_COMPAT") if otel_compat_env is not None: config_data["otel_compat"] = _parse_config_bool( otel_compat_env, field_name="NOVEUM_OTEL_COMPAT" ) pii_enabled_env = os.getenv("NOVEUM_PII_ENABLED") if pii_enabled_env is not None: if "security" not in config_data: config_data["security"] = {} config_data["security"]["pii_enabled"] = _parse_config_bool( pii_enabled_env, field_name="NOVEUM_PII_ENABLED", ) pii_salt_env = os.getenv("NOVEUM_PII_SALT") if pii_salt_env is not None: if "security" not in config_data: config_data["security"] = {} config_data["security"]["pii_salt"] = pii_salt_env dev_traces_dir_env = os.getenv("NOVEUM_DEV_TRACES_DIR") if dev_traces_dir_env: config_data["dev_traces_dir"] = dev_traces_dir_env # Try to load from config files config_files = [ "noveum-trace.yaml", "noveum-trace.yml", "noveum-trace.json", ".noveum-trace.yaml", ".noveum-trace.yml", ".noveum-trace.json", ] for config_file in config_files: if os.path.exists(config_file): file_config = _load_from_file(config_file) # Merge file config with environment config (env takes precedence). # Deep-merge nested dicts (e.g. security) so env-only keys like # NOVEUM_PII_ENABLED do not wipe file-provided security.pii_salt. config_data = _deep_merge_dicts(file_config.to_dict(), config_data) break return Config.from_dict(config_data) def _load_from_file(file_path: Union[str, Path]) -> Config: """Load configuration from a file.""" file_path = Path(file_path) if not file_path.exists(): raise ConfigurationError(f"Configuration file not found: {file_path}") try: with open(file_path) as f: if file_path.suffix in (".yaml", ".yml"): data = yaml.safe_load(f) elif file_path.suffix == ".json": data = json.load(f) else: raise ConfigurationError( f"Unsupported configuration file format: {file_path.suffix}. " "Supported formats: .yaml, .yml, .json" ) return Config.from_dict(data or {}) except (yaml.YAMLError, json.JSONDecodeError) as e: raise ConfigurationError( f"Failed to parse configuration file {file_path}: {e}" ) from e except Exception as e: raise ConfigurationError( f"Failed to load configuration file {file_path}: {e}" ) from e