core

Core module - business logic, models, and utilities.

ConfigurationError

ConfigurationError(message, config_path=None)

Bases: Exception

Raised when configuration loading or parsing fails.

Initialize the configuration error.

Parameters:

Name	Type	Description	Default
message	str	Error description	required
config_path	str | None	Path to the config file that caused the error	None

Source code in src/core/core_config.py

def __init__(self, message: str, config_path: str | None = None) -> None:
    """Initialize the configuration error.

    Args:
        message: Error description
        config_path: Path to the config file that caused the error

    """
    super().__init__(message)
    self.config_path = config_path

load_config

load_config(config_path)

Load the configuration from a YAML file, resolve environment variables, and validate it.

Parameters:

Name	Type	Description	Default
config_path	str	Path to the configuration YAML file.	required

Returns:

Type	Description
AppConfig	Validated AppConfig Pydantic model with resolved env vars.

Raises:

Type	Description
FileNotFoundError	If the config file does not exist.
ValueError	If the configuration is invalid, the path is insecure, or env vars are missing.
PermissionError	If the config file cannot be read.
YAMLError	If there is an error parsing the YAML file.
RuntimeError	For unexpected errors during additional validation steps.

Source code in src/core/core_config.py

def load_config(config_path: str) -> AppConfig:
    """Load the configuration from a YAML file, resolve environment variables, and validate it.

    Args:
        config_path: Path to the configuration YAML file.

    Returns:
        Validated AppConfig Pydantic model with resolved env vars.

    Raises:
        FileNotFoundError: If the config file does not exist.
        ValueError: If the configuration is invalid, the path is insecure, or env vars are missing.
        PermissionError: If the config file cannot be read.
        YAMLError: If there is an error parsing the YAML file.
        RuntimeError: For unexpected errors during additional validation steps.

    """
    env_loaded = load_dotenv()
    logger.info(".env file %s", "found and loaded" if env_loaded else "not found, using system environment variables")

    if missing_vars := validate_required_env_vars():
        error_msg = f"Missing required environment variables: {', '.join(missing_vars)}"
        logger.critical(error_msg)
        raise ValueError(error_msg)

    try:
        validated_path = _validate_config_path(config_path)
        config_data = _read_and_parse_config(validated_path)

        logger.debug(
            "[CONFIG] Raw config (before env var resolution):\n%s",
            yaml.dump(config_data),
        )
        config_data = resolve_env_vars(config_data)
        logger.debug(
            "[CONFIG] Resolved config (after env var resolution):\n%s",
            yaml.dump(config_data),
        )

        config_data = _validate_config_data_type(config_data)

        # Validate with Pydantic
        try:
            config_model = AppConfig(**config_data)
        except ValidationError as e:
            error_details = format_pydantic_errors(e)
            msg = f"Configuration validation failed:\n{error_details}"
            raise ValueError(msg) from e

        logger.info("Configuration successfully loaded and validated.")
        return config_model

    except (FileNotFoundError, PermissionError, ValueError, yaml.YAMLError) as e:
        logger.critical("Configuration loading failed: %s", e)
        raise
    except Exception as e:
        logger.critical("An unexpected error occurred during config loading: %s", e, exc_info=True)
        msg = f"An unexpected error occurred during config loading: {e}"
        raise RuntimeError(msg) from e

get_full_log_path

get_full_log_path(config, key, default, error_logger=None)

Return the full log path by joining the base logs directory with the relative path.

Ensures the base directory and the log file's directory exist.

Source code in src/core/logger.py

def get_full_log_path(
    config: AppConfig | None,
    key: str,
    default: str,
    error_logger: logging.Logger | None = None,
) -> str:
    """Return the full log path by joining the base logs directory with the relative path.

    Ensures the base directory and the log file's directory exist.
    """
    logs_base_dir = ""
    relative_path = default

    if config is not None:
        logs_base_dir = config.logs_base_dir
        relative_path = get_path_from_config(config, key, default, error_logger)
    elif error_logger is not None:
        error_logger.error("Invalid config passed to get_full_log_path.")

    # Ensure the base directory exists
    ensure_directory(logs_base_dir, error_logger)

    # Join base directory and relative path
    full_path = str(Path(logs_base_dir) / relative_path)

    if log_dir := str(Path(full_path).parent):
        ensure_directory(log_dir, error_logger)

    return full_path

get_loggers

get_loggers(config)

Create and return loggers with QueueHandler for non-blocking file logging.

Sets up multiple loggers with different purposes and a shared queue listener for efficient file I/O. Uses RichHandler for colorized console output.

Note

This function never raises exceptions. On setup failure, it returns fallback loggers with basic StreamHandler configuration.

Parameters:

Name	Type	Description	Default
config	AppConfig	Typed application configuration.	required

Returns:

Type	Description
Logger	Tuple of (console_logger, error_logger, analytics_logger,
Logger	db_verify_logger, listener). Listener is None on failure.

Source code in src/core/logger.py

def get_loggers(
    config: AppConfig,
) -> tuple[logging.Logger, logging.Logger, logging.Logger, logging.Logger, SafeQueueListener | None]:
    """Create and return loggers with QueueHandler for non-blocking file logging.

    Sets up multiple loggers with different purposes and a shared queue listener
    for efficient file I/O. Uses RichHandler for colorized console output.

    Note:
        This function never raises exceptions. On setup failure, it returns
        fallback loggers with basic StreamHandler configuration.

    Args:
        config: Typed application configuration.

    Returns:
        Tuple of (console_logger, error_logger, analytics_logger,
        db_verify_logger, listener). Listener is ``None`` on failure.

    """
    try:
        levels = get_log_levels_from_config(config)
        log_files = get_log_file_paths(config)
        console_logger = create_console_logger(levels)

        _, error_logger, analytics_logger, db_verify_logger, listener = setup_queue_logging(config, levels, log_files)

    except (ImportError, OSError, ValueError, AttributeError, TypeError) as e:
        return create_fallback_loggers(e)

    console_logger.debug("Logging setup with QueueListener and RichHandler complete.")
    return console_logger, error_logger, analytics_logger, db_verify_logger, listener