Coverage for src/qdrant_loader/config/__init__.py: 86%
201 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Configuration module.
3This module provides the main configuration interface for the application.
4It combines global settings with source-specific configurations.
5"""
7import os
8import re
9from pathlib import Path
10from typing import Any, Optional
12import yaml
13from dotenv import load_dotenv
14from pydantic import Field, ValidationError, model_validator
15from pydantic_settings import BaseSettings, SettingsConfigDict
17from ..utils.logging import LoggingConfig
18from ..utils.sensitive import sanitize_exception_message
19from .chunking import ChunkingConfig
20from .concurrency import ConcurrencyConfig
22# Import consolidated configs
23from .global_config import GlobalConfig, SemanticAnalysisConfig
25# Import multi-project support
26from .models import (
27 ParsedConfig,
28 ProjectConfig,
29 ProjectContext,
30 ProjectDetail,
31 ProjectInfo,
32 ProjectsConfig,
33 ProjectStats,
34)
35from .parser import MultiProjectConfigParser
36from .sources import SourcesConfig
37from .state import StateManagementConfig
38from .validator import ConfigValidator
39from .workspace import WorkspaceConfig
41# Load environment variables from .env file
42load_dotenv(override=False)
45def _get_logger():
46 return LoggingConfig.get_logger(__name__)
49# Lazy import function for connector configs
50def _get_connector_configs():
51 """Lazy import connector configs to avoid circular dependencies."""
52 from ..connectors.confluence.config import ConfluenceSpaceConfig
53 from ..connectors.git.config import GitAuthConfig, GitRepoConfig
54 from ..connectors.jira.config import JiraProjectConfig
55 from ..connectors.publicdocs.config import PublicDocsSourceConfig, SelectorsConfig
57 return {
58 "ConfluenceSpaceConfig": ConfluenceSpaceConfig,
59 "GitAuthConfig": GitAuthConfig,
60 "GitRepoConfig": GitRepoConfig,
61 "JiraProjectConfig": JiraProjectConfig,
62 "PublicDocsSourceConfig": PublicDocsSourceConfig,
63 "SelectorsConfig": SelectorsConfig,
64 }
67__all__ = [
68 "ChunkingConfig",
69 "ConcurrencyConfig",
70 "ConfluenceSpaceConfig",
71 "GitAuthConfig",
72 "GitRepoConfig",
73 "GlobalConfig",
74 "JiraProjectConfig",
75 "PublicDocsSourceConfig",
76 "SelectorsConfig",
77 "SemanticAnalysisConfig",
78 "Settings",
79 "SourcesConfig",
80 "StateManagementConfig",
81 # Multi-project support
82 "ProjectContext",
83 "ProjectConfig",
84 "ProjectsConfig",
85 "ParsedConfig",
86 "ProjectStats",
87 "ProjectInfo",
88 "ProjectDetail",
89 "MultiProjectConfigParser",
90 "ConfigValidator",
91 # Functions
92 "get_global_config",
93 "get_settings",
94 "initialize_config",
95 "initialize_config_with_workspace",
96]
99# Add lazy loading for connector configs
100def __getattr__(name):
101 """Lazy import connector configs to avoid circular dependencies."""
102 connector_configs = _get_connector_configs()
103 if name in connector_configs:
104 return connector_configs[name]
105 raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
108_global_settings: Optional["Settings"] = None
111def get_settings() -> "Settings":
112 """Get the global settings instance.
114 Returns:
115 Settings: The global settings instance.
116 """
117 if _global_settings is None:
118 raise RuntimeError(
119 "Settings not initialized. Call initialize_config() or initialize_config_with_workspace() first."
120 )
121 return _global_settings
124def get_global_config() -> GlobalConfig:
125 """Get the global configuration instance.
127 Returns:
128 GlobalConfig: The global configuration instance.
129 """
130 return get_settings().global_config
133def initialize_config(
134 yaml_path: Path, env_path: Path | None = None, skip_validation: bool = False
135) -> None:
136 """Initialize the global configuration.
138 Args:
139 yaml_path: Path to the YAML configuration file.
140 env_path: Optional path to the .env file.
141 skip_validation: If True, skip directory validation and creation.
142 """
143 global _global_settings
144 try:
145 # Proceed with initialization
146 _get_logger().debug(
147 "Initializing configuration",
148 yaml_path=str(yaml_path),
149 env_path=str(env_path) if env_path else None,
150 )
151 _global_settings = Settings.from_yaml(
152 yaml_path, env_path=env_path, skip_validation=skip_validation
153 )
154 _get_logger().debug("Successfully initialized configuration")
156 except Exception as e:
157 safe_error = sanitize_exception_message(e)
158 _get_logger().error(
159 "Failed to initialize configuration",
160 error=safe_error,
161 yaml_path=str(yaml_path),
162 )
163 raise
166def initialize_config_with_workspace(
167 workspace_config: WorkspaceConfig, skip_validation: bool = False
168) -> None:
169 """Initialize configuration using workspace settings.
171 Args:
172 workspace_config: Workspace configuration with paths and settings
173 skip_validation: If True, skip directory validation and creation
174 """
175 global _global_settings
176 try:
177 _get_logger().debug(
178 "Initializing configuration with workspace",
179 workspace=str(workspace_config.workspace_path),
180 config_path=str(workspace_config.config_path),
181 env_path=(
182 str(workspace_config.env_path) if workspace_config.env_path else None
183 ),
184 )
186 # Load configuration using workspace paths
187 _global_settings = Settings.from_yaml(
188 workspace_config.config_path,
189 env_path=workspace_config.env_path,
190 skip_validation=skip_validation,
191 )
193 # Check if database_path was specified in config.yaml and warn user
194 original_db_path = _global_settings.global_config.state_management.database_path
195 workspace_db_path = str(workspace_config.database_path)
197 # Only warn if the original path is different from the workspace path and not empty/default
198 if (
199 original_db_path
200 and original_db_path != ":memory:"
201 and original_db_path != workspace_db_path
202 ):
203 _get_logger().warning(
204 "Database path in config.yaml is ignored in workspace mode",
205 config_database_path=original_db_path,
206 workspace_database_path=workspace_db_path,
207 )
209 # Override the database path with workspace-specific path
210 _global_settings.global_config.state_management.database_path = (
211 workspace_db_path
212 )
214 _get_logger().debug(
215 "Set workspace database path",
216 database_path=workspace_db_path,
217 )
219 _get_logger().debug(
220 "Successfully initialized configuration with workspace",
221 workspace=str(workspace_config.workspace_path),
222 )
224 except Exception as e:
225 safe_error = sanitize_exception_message(e)
226 _get_logger().error(
227 "Failed to initialize configuration with workspace",
228 error=safe_error,
229 workspace=str(workspace_config.workspace_path),
230 )
231 raise
234class Settings(BaseSettings):
235 """Main configuration class combining global and source-specific settings."""
237 # Configuration objects - these are the only fields we need
238 global_config: GlobalConfig = Field(
239 default_factory=GlobalConfig, description="Global configuration settings"
240 )
241 projects_config: ProjectsConfig = Field(
242 default_factory=ProjectsConfig, description="Multi-project configurations"
243 )
245 model_config = SettingsConfigDict(
246 env_file=None, # Disable automatic .env loading - we handle this manually
247 env_file_encoding="utf-8",
248 extra="allow",
249 )
251 @model_validator(mode="after") # type: ignore
252 def validate_source_configs(self) -> "Settings":
253 """Validate that required configuration is present for configured sources."""
254 _get_logger().debug("Validating source configurations")
256 # Auto-resolve environment variables as fallbacks
257 self._auto_resolve_env_vars()
259 # Validate that required fields are not empty after variable substitution
260 if not self.global_config.qdrant.url:
261 raise ValueError(
262 "Qdrant URL is required but was not provided or substituted"
263 )
265 if not self.global_config.qdrant.collection_name:
266 raise ValueError(
267 "Qdrant collection name is required but was not provided or substituted"
268 )
270 # Note: Source validation is now handled at the project level
271 # Each project's sources are validated when the project is processed
273 _get_logger().debug("Source configuration validation successful")
274 return self
276 def _auto_resolve_env_vars(self) -> None:
277 """Auto-resolve well-known environment variables as fallbacks.
279 Priority: config file value > environment variable > default.
280 Only fills in values that were not explicitly set in config.
282 Note: detection uses default-value sentinels, so explicitly setting a
283 config value equal to the default (e.g. url: http://localhost:6333)
284 will still be overridden by the environment variable.
285 """
286 # OPENAI_API_KEY → embedding.api_key and llm.api_key
287 openai_key = os.getenv("OPENAI_API_KEY")
288 if openai_key:
289 if not self.global_config.embedding.api_key:
290 self.global_config.embedding.api_key = openai_key
291 if self.global_config.llm and isinstance(self.global_config.llm, dict):
292 if not self.global_config.llm.get("api_key"):
293 self.global_config.llm["api_key"] = openai_key
295 # QDRANT_URL → qdrant.url (override only if still default)
296 qdrant_url = os.getenv("QDRANT_URL")
297 if qdrant_url and self.global_config.qdrant.url == "http://localhost:6333":
298 self.global_config.qdrant.url = qdrant_url
300 # QDRANT_API_KEY → qdrant.api_key
301 qdrant_api_key = os.getenv("QDRANT_API_KEY")
302 if qdrant_api_key and not self.global_config.qdrant.api_key:
303 self.global_config.qdrant.api_key = qdrant_api_key
305 # QDRANT_COLLECTION_NAME → qdrant.collection_name
306 collection = os.getenv("QDRANT_COLLECTION_NAME")
307 if collection and self.global_config.qdrant.collection_name == "documents":
308 self.global_config.qdrant.collection_name = collection
310 # STATE_DB_PATH → state_management.database_path
311 # Note: In workspace mode this is overridden by workspace_config.database_path
312 state_db = os.getenv("STATE_DB_PATH")
313 if (
314 state_db
315 and self.global_config.state_management.database_path == "./state.db"
316 ):
317 self.global_config.state_management.database_path = state_db
319 # STATE_DB_URL → state_management.database_url
320 # when set, selects the backend by dialect (e.g., postgresql+asyncpg://user:pass@host:5432/dbname)
321 # and overrides database_path. Namespaced to avoid clashing with a generic
322 # DATABASE_URL that might be used for other purposes.
323 state_database_url = os.getenv("STATE_DB_URL")
324 if state_database_url and not self.global_config.state_management.database_url:
325 self.global_config.state_management.database_url = state_database_url
327 @property
328 def qdrant_url(self) -> str:
329 """Get the Qdrant URL from global configuration."""
330 return self.global_config.qdrant.url
332 @property
333 def qdrant_api_key(self) -> str | None:
334 """Get the Qdrant API key from global configuration."""
335 return self.global_config.qdrant.api_key
337 @property
338 def qdrant_collection_name(self) -> str:
339 """Get the Qdrant collection name from global configuration."""
340 return self.global_config.qdrant.collection_name
342 @property
343 def openai_api_key(self) -> str:
344 """Get the OpenAI API key from embedding configuration."""
345 api_key = self.global_config.embedding.api_key
346 if not api_key:
347 raise ValueError(
348 "OpenAI API key is required but was not provided or substituted in embedding configuration"
349 )
350 return api_key
352 @property
353 def state_db_path(self) -> str:
354 """Get the state database path from global configuration."""
355 return self.global_config.state_management.database_path
357 @property
358 def llm_settings(self):
359 """Provider-agnostic LLM settings derived from global configuration.
361 Uses `global.llm` when present; otherwise maps legacy fields.
362 """
363 # Import lazily to avoid hard dependency issues in environments without core installed
364 from importlib import import_module
366 settings_mod = import_module("qdrant_loader_core.llm.settings")
367 LLMSettings = settings_mod.LLMSettings
368 return LLMSettings.from_global_config(self.global_config.to_dict())
370 @staticmethod
371 def _substitute_env_vars(data: Any) -> Any:
372 """Recursively substitute environment variables in configuration data.
374 Args:
375 data: Configuration data to process
377 Returns:
378 Processed data with environment variables substituted
379 """
380 if isinstance(data, str):
381 # First expand $HOME if present
382 if "$HOME" in data:
383 data = data.replace("$HOME", os.path.expanduser("~"))
385 # Then handle ${VAR_NAME} pattern
386 pattern = r"\${([^}]+)}"
387 matches = re.finditer(pattern, data)
388 result = data
389 for match in matches:
390 var_name = match.group(1)
391 env_value = os.getenv(var_name)
392 if env_value is None:
393 # Only warn about missing variables that are commonly required
394 # Skip STATE_DB_PATH as it's often overridden in workspace mode
395 if var_name not in ["STATE_DB_PATH"]:
396 _get_logger().warning(
397 "Environment variable not found", variable=var_name
398 )
399 continue
400 # If the environment variable contains $HOME, expand it
401 if "$HOME" in env_value:
402 env_value = env_value.replace("$HOME", os.path.expanduser("~"))
403 result = result.replace(f"${{{var_name}}}", env_value)
405 return result
406 elif isinstance(data, dict):
407 return {k: Settings._substitute_env_vars(v) for k, v in data.items()}
408 elif isinstance(data, list):
409 return [Settings._substitute_env_vars(item) for item in data]
410 return data
412 @staticmethod
413 def _validate_env_substitution(data: Any) -> None:
414 """Validate that all environment variables in config have been substituted.
416 Raises:
417 ValueError: If any ${VAR_NAME} pattern remains in the configuration.
418 """
419 pattern = r"\$\{([^}]+)\}"
421 def check_value(value: Any, path: str = "") -> None:
422 if isinstance(value, str):
423 matches = re.finditer(pattern, value)
424 for match in matches:
425 var_name = match.group(1)
426 location = f" at {path}" if path else ""
427 raise ValueError(
428 f"Environment variable {var_name!r} not found{location}. "
429 f"Please set {var_name} in your environment or .env file, "
430 f"or comment out the configuration line that references it."
431 )
432 elif isinstance(value, dict):
433 for k, v in value.items():
434 check_value(v, f"{path}.{k}" if path else k)
435 elif isinstance(value, list):
436 for i, item in enumerate(value):
437 check_value(item, f"{path}[{i}]")
439 check_value(data)
441 @classmethod
442 def from_yaml(
443 cls,
444 config_path: Path,
445 env_path: Path | None = None,
446 skip_validation: bool = False,
447 ) -> "Settings":
448 """Load configuration from a YAML file.
450 Args:
451 config_path: Path to the YAML configuration file.
452 env_path: Optional path to the .env file. If provided, only this file is loaded.
453 skip_validation: If True, skip directory validation and creation.
455 Returns:
456 Settings: Loaded configuration.
457 """
458 _get_logger().debug("Loading configuration from YAML", path=str(config_path))
459 try:
460 # Step 1: Load environment variables first
461 if env_path is not None:
462 # Custom env file specified - load only this file
463 _get_logger().debug(
464 "Loading custom environment file", path=str(env_path)
465 )
466 if not env_path.exists():
467 raise FileNotFoundError(f"Environment file not found: {env_path}")
468 load_dotenv(env_path, override=True)
469 else:
470 # Load default .env file if it exists
471 _get_logger().debug("Loading default environment variables")
472 load_dotenv(override=False)
474 # Step 2: Load YAML config
475 with open(config_path) as f:
476 config_data = yaml.safe_load(f)
478 # Step 3: Process all environment variables in config using substitution
479 _get_logger().debug("Processing environment variables in configuration")
480 config_data = cls._substitute_env_vars(config_data)
482 # Step 3.5: In strict mode, fail fast if placeholders remain unresolved.
483 # Tests and template flows often use skip_validation=True and should keep
484 # legacy behavior where unresolved placeholders can remain in config.
485 if not skip_validation:
486 cls._validate_env_substitution(config_data)
488 # Step 4: Use multi-project parser to parse configuration
489 validator = ConfigValidator()
490 parser = MultiProjectConfigParser(validator)
491 parsed_config = parser.parse(config_data, skip_validation=skip_validation)
493 # Step 5: Create settings instance with parsed configuration
494 settings = cls(
495 global_config=parsed_config.global_config,
496 projects_config=parsed_config.projects_config,
497 )
499 _get_logger().debug("Successfully created Settings instance")
500 return settings
502 except yaml.YAMLError as e:
503 _get_logger().error(
504 "Failed to parse YAML configuration",
505 error=sanitize_exception_message(e),
506 )
507 raise
508 except ValidationError as e:
509 _get_logger().error(
510 "Configuration validation failed",
511 error=sanitize_exception_message(e),
512 )
513 raise
514 except Exception as e:
515 _get_logger().error(
516 "Unexpected error loading configuration",
517 error=sanitize_exception_message(e),
518 )
519 raise
521 def to_dict(self) -> dict:
522 """Convert the configuration to a dictionary.
524 Returns:
525 dict: Configuration as a dictionary.
526 """
527 return {
528 "global": self.global_config.to_dict(),
529 "projects": self.projects_config.to_dict(),
530 }