Config

churn_prediction.config.settings

Application Settings and Configuration Management.

This module defines the type-safe application configuration schema using Pydantic Settings and Pydantic v2. It manages environment variables, .env file overrides, runtime directories, and hyperparameter configuration (params.yaml).

Environment

Bases: StrEnum

Enumeration of supported application logging levels.

EvaluateParams

Bases: ImmutableBaseModel

Define model evaluation and classification configuration.

Contains the decision threshold and other settings used to evaluate model performance on the validation dataset.

ImmutableBaseModel

Bases: BaseModel

Base Pydantic model enforcing immutable and strict configuration.

Subclasses inherit validation behavior intended to prevent accidental mutation of application configuration after initialization.

LogLevel

Bases: StrEnum

Enumeration of logging levels.

MetadataParams

Bases: ImmutableBaseModel

Define metadata used for model tracking and registry records.

PipelineParams

Bases: ImmutableBaseModel

Define the complete validated pipeline configuration.

Aggregates dataset schema, preprocessing, training, evaluation, and metadata configuration into a single type-safe parameter object.

load_from_yaml(path=DEFAULT_PARAMS_FILE) classmethod

Load and validate pipeline parameters from a YAML configuration file.

The YAML contents are parsed into the validated parameter model. If the configuration file does not exist, the model's default configuration is returned instead.

Parameters:

Name	Type	Description	Default
path	Path	Path to the YAML parameter configuration file.	DEFAULT_PARAMS_FILE

Returns:

Type	Description
PipelineParams	Validated pipeline parameter configuration.

Raises:

Type	Description
ValueError	If the YAML file exists but contains invalid parameter values or cannot be parsed into the expected configuration schema.

Source code in src/churn_prediction/config/settings.py

@classmethod
def load_from_yaml(cls, path: Path = DEFAULT_PARAMS_FILE) -> "PipelineParams":
    """Load and validate pipeline parameters from a YAML configuration file.

    The YAML contents are parsed into the validated parameter model. If the
    configuration file does not exist, the model's default configuration is
    returned instead.

    Parameters
    ----------
    path : pathlib.Path, default=DEFAULT_PARAMS_FILE
        Path to the YAML parameter configuration file.

    Returns
    -------
    PipelineParams
        Validated pipeline parameter configuration.

    Raises
    ------
    ValueError
        If the YAML file exists but contains invalid parameter values or
        cannot be parsed into the expected configuration schema.
    """
    if not path.is_file():
        return cls()

    try:
        with path.open(encoding="utf-8") as f:
            raw_data: dict[str, Any] = yaml.safe_load(f) or {}
        return cls.model_validate(raw_data)
    except yaml.YAMLError as exc:
        raise RuntimeError(f"Malformed YAML configuration in '{path}': {exc}") from exc
    except ValidationError as exc:
        raise RuntimeError(f"Invalid pipeline parameter schema in '{path}': {exc}") from exc

PreprocessingParams

Bases: ImmutableBaseModel

Define preprocessing configuration for model-ready datasets.

Contains the feature transformation settings used consistently during training and inference.

SchemaParams

Bases: ImmutableBaseModel

Define dataset schema and feature configuration.

Stores the column definitions and feature groups used to validate, preprocess, and model the churn prediction datasets.

Settings

Bases: BaseSettings

Define validated application-wide runtime configuration.

Configuration values are loaded from environment variables and the project-level .env file, with environment variables taking precedence.

The settings object contains runtime paths, logging configuration, external service credentials, API configuration, model settings, and computed artifact locations.

Notes

Unknown configuration values are ignored, and environment variable names are matched case-insensitively.

PARAMS property

Return the lazily loaded pipeline configuration.

Loads and validates params.yaml on first access and caches the resulting configuration for subsequent accesses.

Returns:

Type	Description
PipelineParams	Validated pipeline hyperparameters and processing configuration.

assemble_api_urls()

Construct API endpoint URLs from the configured host and port.

Derives the URLs used by the prediction, download, explainability, and health-check clients from the application's API host and port settings.

Source code in src/churn_prediction/config/settings.py

@model_validator(mode="after")
def assemble_api_urls(self) -> "Settings":
    """Construct API endpoint URLs from the configured host and port.

    Derives the URLs used by the prediction, download, explainability, and
    health-check clients from the application's API host and port settings.
    """
    base_url = self.API_BASE_URL or f"http://{self.FASTAPI_HOST}:{self.FASTAPI_PORT}"
    self.API_BASE_URL = base_url.rstrip("/")
    self.PREDICT_API_URL = f"{self.API_BASE_URL}/api/v1/predict/batch"
    self.DOWNLOAD_API_URL = f"{self.API_BASE_URL}/api/v1/predictions/download"
    self.EXPLAIN_API_URL = f"{self.API_BASE_URL}/explain"
    self.HEALTH_API_URL = f"{self.API_BASE_URL}/health"

    return self

ensure_directories()

Create required application workspace directories.

Ensures that directories required for data ingestion, preprocessing, models, reports, predictions, logs, and other runtime artifacts exist. Existing directories are left unchanged.

Source code in src/churn_prediction/config/settings.py

def ensure_directories(self) -> None:
    """Create required application workspace directories.

    Ensures that directories required for data ingestion, preprocessing,
    models, reports, predictions, logs, and other runtime artifacts exist.
    Existing directories are left unchanged.
    """
    directories = [
        self.LOGS_DIR,
        self.RAW_DATA_DIR,
        self.PROCESSED_DATA_DIR,
        self.PREDICTION_DATA_DIR,
        self.MODEL_DIR,
        self.VALIDATION_REPORT_DIR,
        self.TRAINING_REPORT_DIR,
        self.SHAP_REPORT_DIR,
    ]
    for directory in directories:
        directory.mkdir(parents=True, exist_ok=True)

TrainParams

Bases: ImmutableBaseModel

Define model training configuration.

Stores XGBoost parameters and training controls such as boosting rounds, early stopping, and evaluation verbosity.

XGBoostParams

Bases: ImmutableBaseModel

Define hyperparameters used to configure the XGBoost model.

get_settings() cached

Return the cached application settings.

Local development may use a project-level environment file, while deployment environments such as Render provide configuration directly through environment variables.

Source code in src/churn_prediction/config/settings.py

@lru_cache
def get_settings() -> Settings:
    """Return the cached application settings.

    Local development may use a project-level environment file, while
    deployment environments such as Render provide configuration directly
    through environment variables.
    """
    settings = Settings()
    settings.ensure_directories()

    return settings

churn_prediction.config.logger

Structured Application Logging Module.

Implements a centralized logging system that provides formatted console output and rotating file logs. It is designed to be thread-safe and prevents duplicate handler initialization in concurrent environments.

LoggingLifecycle

Track the process-local state of logging configuration.

This internal helper provides thread-safe state management for the application logging lifecycle. It prevents multiple concurrent callers from configuring the logging system more than once.

Attributes:

Name	Type	Description
_lock	Lock	Lock protecting logging initialization from concurrent access.
_is_configured	bool	Whether application-wide logging has been successfully configured.

is_configured() classmethod

Return whether application logging has been configured.

Returns:

Type	Description
bool	True when logging configuration has completed successfully; otherwise False.

Source code in src/churn_prediction/config/logger.py

@classmethod
def is_configured(cls) -> bool:
    """Return whether application logging has been configured.

    Returns
    -------
    bool
        ``True`` when logging configuration has completed successfully;
        otherwise ``False``.
    """
    return cls._is_configured

mark_configured() classmethod

Mark application logging as successfully configured.

Source code in src/churn_prediction/config/logger.py

@classmethod
def mark_configured(cls) -> None:
    """Mark application logging as successfully configured."""
    cls._is_configured = True

configure_logging(*, force=False)

Configure application-wide structured logging.

Initializes Structlog, console output, rotating file logging, third-party logger handling, standard warning capture, and optional Sentry error monitoring.

Configuration is idempotent by default. If logging has already been configured, subsequent calls have no effect unless force=True is specified.

Parameters:

Name	Type	Description	Default
force	bool	Whether to rebuild the logging configuration when logging has already been initialized. When True, existing root handlers are removed before configuration is recreated.	False

Source code in src/churn_prediction/config/logger.py

def configure_logging(*, force: bool = False) -> None:
    """Configure application-wide structured logging.

    Initializes Structlog, console output, rotating file logging,
    third-party logger handling, standard warning capture, and optional
    Sentry error monitoring.

    Configuration is idempotent by default. If logging has already been
    configured, subsequent calls have no effect unless ``force=True`` is
    specified.

    Parameters
    ----------
    force : bool, default=False
        Whether to rebuild the logging configuration when logging has already
        been initialized. When ``True``, existing root handlers are removed
        before configuration is recreated.
    """
    if LoggingLifecycle.is_configured() and not force:
        return

    with LoggingLifecycle._lock:
        if LoggingLifecycle.is_configured() and not force:
            return

        # Initialize the settings and determine the environment
        settings = get_settings()
        is_sentry_environment = settings.ENVIRONMENT.value in {"STAGING", "PRODUCTION"}

        # 0. Initialize Sentry
        if settings.SENTRY_DSN and is_sentry_environment:
            _init_sentry(
                str(settings.SENTRY_DSN),
                settings.ENVIRONMENT.value,
            )

        shared_processors = _get_shared_processors()

        # 1. Configure structlog engine
        structlog.configure(
            processors=[
                *shared_processors,
                structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
            ],
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
            cache_logger_on_first_use=True,
        )

        # 2. Choose renderer setup
        console_renderer = (
            structlog.processors.JSONRenderer()
            if is_sentry_environment
            else structlog.dev.ConsoleRenderer(colors=True)
        )

        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(
            structlog.stdlib.ProcessorFormatter(
                foreign_pre_chain=shared_processors,
                processors=[
                    structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                    console_renderer,
                ],
            )
        )

        # 3. Configure rotating file handler setup
        file_handler = logging.handlers.TimedRotatingFileHandler(
            filename=str(
                settings.LOGS_DIR
                / (f"{datetime.now(UTC):%Y%m%dT%H%M%S}_{uuid4().hex[:4]}" + ".log")
            ),
            when=settings.LOG_ROTATION_WHEN,
            interval=1,
            backupCount=settings.LOG_BACKUP_COUNT,
            encoding="utf-8",
        )
        file_handler.setFormatter(
            structlog.stdlib.ProcessorFormatter(
                foreign_pre_chain=shared_processors,
                processors=[
                    structlog.stdlib.ProcessorFormatter.remove_processors_meta,
                    structlog.processors.JSONRenderer(),
                ],
            )
        )

        # 4. Attach handlers to root logger & clear default handlers
        root_logger = logging.getLogger()
        root_logger.handlers.clear()

        root_logger.addHandler(console_handler)
        root_logger.addHandler(file_handler)
        root_logger.setLevel(settings.LOG_LEVEL.value.upper())

        # 5. Clear handlers for third-party loggers to avoid duplicate logs
        for logger_name in ("uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"):
            third_party_logger = logging.getLogger(logger_name)
            third_party_logger.handlers.clear()
            third_party_logger.propagate = True

        # Redirect standard library warnings to logging ecosystem
        logging.captureWarnings(capture=True)

        LoggingLifecycle.mark_configured()

get_logger(name=None)

Return a configured application logger.

Logging configuration is initialized lazily if it has not already been configured.

Parameters:

Name	Type	Description	Default
name	str or None	Logger name to bind to the returned logger. If None, the default application logger name configured in :class:Settings is used.	None

Returns:

Type	Description
BoundLogger	Configured Structlog logger suitable for structured application logging.

Source code in src/churn_prediction/config/logger.py

def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger:
    """Return a configured application logger.

    Logging configuration is initialized lazily if it has not already been
    configured.

    Parameters
    ----------
    name : str or None, default=None
        Logger name to bind to the returned logger. If ``None``, the default
        application logger name configured in :class:`Settings` is used.

    Returns
    -------
    structlog.stdlib.BoundLogger
        Configured Structlog logger suitable for structured application
        logging.
    """
    if not LoggingLifecycle.is_configured():
        configure_logging()

    settings = get_settings()
    logger_name = name or settings.LOGGER_NAME
    return structlog.get_logger(logger_name)