53e06fff63
- Add utils/config.py module using python-dotenv for centralized config - Externalize Box folder IDs, paths, and timeout settings to .env files - Create .env.example template with all configuration variables - Add separate systemd service files for prod and dev environments - Update ford_qc_box_hotfolder_process.py to use config module - Update qc_engine.py and path_resolver.py with optional config support - Maintain backwards compatibility with hardcoded defaults - Update documentation in CLAUDE.md 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
271 lines
9.1 KiB
Python
271 lines
9.1 KiB
Python
"""
|
|
Configuration management module for Ford BnP QC system.
|
|
|
|
Loads configuration from environment variables using python-dotenv,
|
|
with validation for required values and type conversion for numeric values.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from typing import Optional
|
|
from dotenv import load_dotenv
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConfigurationError(Exception):
|
|
"""Raised when required configuration is missing or invalid."""
|
|
pass
|
|
|
|
|
|
class Config:
|
|
"""
|
|
Centralized configuration class for Ford BnP QC system.
|
|
|
|
Loads environment variables from .env files with the following precedence:
|
|
1. System environment variables (highest priority)
|
|
2. .env file in script directory
|
|
3. .env.{FORD_QC_ENV} file in script directory
|
|
|
|
Usage:
|
|
from utils.config import config
|
|
folder_id = config.BOX_SOURCE_FOLDER_ID
|
|
"""
|
|
|
|
_instance: Optional['Config'] = None
|
|
_loaded: bool = False
|
|
|
|
def __new__(cls) -> 'Config':
|
|
"""Singleton pattern to ensure configuration is loaded once."""
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
"""Initialize configuration (only runs once due to singleton)."""
|
|
if not Config._loaded:
|
|
self._load_environment()
|
|
self._validate_required()
|
|
Config._loaded = True
|
|
|
|
def _load_environment(self) -> None:
|
|
"""
|
|
Load environment variables from .env files.
|
|
|
|
Searches for .env files in the following order:
|
|
1. Script directory (detected automatically)
|
|
2. Current working directory
|
|
"""
|
|
# Determine script directory
|
|
script_dir = os.environ.get('FORD_QC_SCRIPT_DIR')
|
|
if not script_dir:
|
|
# Auto-detect based on this file's location (utils/config.py -> parent dir)
|
|
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Load environment-specific .env file first (if FORD_QC_ENV is set via system env)
|
|
env_name = os.environ.get('FORD_QC_ENV', '').lower()
|
|
if env_name:
|
|
env_specific_file = os.path.join(script_dir, f'.env.{env_name}')
|
|
if os.path.exists(env_specific_file):
|
|
load_dotenv(env_specific_file, override=False)
|
|
logger.debug(f"Loaded environment-specific config: {env_specific_file}")
|
|
|
|
# Load base .env file (lower priority - won't override env-specific)
|
|
env_file = os.path.join(script_dir, '.env')
|
|
if os.path.exists(env_file):
|
|
load_dotenv(env_file, override=False)
|
|
logger.debug(f"Loaded configuration from: {env_file}")
|
|
|
|
# Store resolved script directory
|
|
self._script_dir = script_dir
|
|
|
|
def _validate_required(self) -> None:
|
|
"""
|
|
Validate that all required configuration values are present.
|
|
|
|
Raises:
|
|
ConfigurationError: If required values are missing.
|
|
"""
|
|
required_vars = [
|
|
'BOX_SOURCE_FOLDER_ID',
|
|
'BOX_REPORT_FOLDER_ID',
|
|
'BOX_PROCESSED_FOLDER_ID',
|
|
'BOX_CONNECTION_TIMEOUT',
|
|
'BOX_READ_TIMEOUT',
|
|
'MAX_RETRIES',
|
|
'RETRY_BACKOFF_BASE',
|
|
'WATCHDOG_INTERVAL',
|
|
]
|
|
|
|
missing = []
|
|
for var in required_vars:
|
|
if not os.environ.get(var):
|
|
missing.append(var)
|
|
|
|
if missing:
|
|
raise ConfigurationError(
|
|
f"Missing required configuration variables: {', '.join(missing)}\n"
|
|
f"Please ensure these are set in your .env file or environment.\n"
|
|
f"See .env.example for reference."
|
|
)
|
|
|
|
def _get_path(self, env_var: str, default: str) -> str:
|
|
"""
|
|
Get a path from environment, resolving relative paths against script_dir.
|
|
|
|
:param env_var: Environment variable name
|
|
:param default: Default value if not set
|
|
:return: Absolute path
|
|
"""
|
|
value = os.environ.get(env_var, default)
|
|
if not os.path.isabs(value):
|
|
return os.path.join(self._script_dir, value)
|
|
return value
|
|
|
|
def _get_int(self, env_var: str, default: int) -> int:
|
|
"""
|
|
Get an integer from environment with validation.
|
|
|
|
:param env_var: Environment variable name
|
|
:param default: Default value if not set
|
|
:return: Integer value
|
|
:raises ConfigurationError: If value cannot be converted to int
|
|
"""
|
|
value = os.environ.get(env_var)
|
|
if value is None:
|
|
return default
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
raise ConfigurationError(
|
|
f"Configuration variable {env_var} must be an integer, got: {value}"
|
|
)
|
|
|
|
# =========================================================================
|
|
# Environment Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def ENVIRONMENT(self) -> str:
|
|
"""Current environment name (production/development)."""
|
|
return os.environ.get('FORD_QC_ENV', 'production')
|
|
|
|
@property
|
|
def IS_PRODUCTION(self) -> bool:
|
|
"""True if running in production environment."""
|
|
return self.ENVIRONMENT.lower() == 'production'
|
|
|
|
@property
|
|
def IS_DEVELOPMENT(self) -> bool:
|
|
"""True if running in development environment."""
|
|
return self.ENVIRONMENT.lower() == 'development'
|
|
|
|
# =========================================================================
|
|
# Box Folder ID Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def BOX_SOURCE_FOLDER_ID(self) -> str:
|
|
"""Box folder ID for incoming asset files."""
|
|
return os.environ.get('BOX_SOURCE_FOLDER_ID', '')
|
|
|
|
@property
|
|
def BOX_REPORT_FOLDER_ID(self) -> str:
|
|
"""Box folder ID for QC report uploads."""
|
|
return os.environ.get('BOX_REPORT_FOLDER_ID', '')
|
|
|
|
@property
|
|
def BOX_PROCESSED_FOLDER_ID(self) -> str:
|
|
"""Box folder ID for successfully processed files."""
|
|
return os.environ.get('BOX_PROCESSED_FOLDER_ID', '')
|
|
|
|
# =========================================================================
|
|
# Path Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def SCRIPT_DIR(self) -> str:
|
|
"""Base directory for the QC system."""
|
|
return self._script_dir
|
|
|
|
@property
|
|
def BOX_CONFIG_PATH(self) -> str:
|
|
"""Full path to Box JWT configuration file."""
|
|
return self._get_path('BOX_CONFIG_FILE', 'ford_box_config.json')
|
|
|
|
@property
|
|
def DOWNLOAD_PATH(self) -> str:
|
|
"""Directory for temporary file downloads."""
|
|
return self._get_path('DOWNLOAD_PATH', 'download_tmp')
|
|
|
|
@property
|
|
def WORKING_DIR(self) -> str:
|
|
"""Working directory for file extraction."""
|
|
return self._get_path('WORKING_DIR', 'working')
|
|
|
|
@property
|
|
def QC_PROFILE_PATH(self) -> str:
|
|
"""Path to the QC profile JSON file."""
|
|
return self._get_path('QC_PROFILE_PATH', 'profiles/ford_bnp.json')
|
|
|
|
@property
|
|
def FALLBACK_WORKING_DIR(self) -> str:
|
|
"""Fallback working directory when primary fails."""
|
|
return os.environ.get(
|
|
'FALLBACK_WORKING_DIR',
|
|
'/home/box-cli/FORD_SCRIPTS/FORD_ASSET_PACK_QC_NEW/working'
|
|
)
|
|
|
|
# =========================================================================
|
|
# Timeout/Retry Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def BOX_CONNECTION_TIMEOUT(self) -> int:
|
|
"""Seconds to wait for Box API connection."""
|
|
return self._get_int('BOX_CONNECTION_TIMEOUT', 30)
|
|
|
|
@property
|
|
def BOX_READ_TIMEOUT(self) -> int:
|
|
"""Seconds to wait for Box API response."""
|
|
return self._get_int('BOX_READ_TIMEOUT', 90)
|
|
|
|
@property
|
|
def MAX_RETRIES(self) -> int:
|
|
"""Maximum retry attempts for failed operations."""
|
|
return self._get_int('MAX_RETRIES', 3)
|
|
|
|
@property
|
|
def RETRY_BACKOFF_BASE(self) -> int:
|
|
"""Base for exponential backoff calculation."""
|
|
return self._get_int('RETRY_BACKOFF_BASE', 2)
|
|
|
|
@property
|
|
def WATCHDOG_INTERVAL(self) -> int:
|
|
"""Seconds between watchdog notifications."""
|
|
return self._get_int('WATCHDOG_INTERVAL', 30)
|
|
|
|
# =========================================================================
|
|
# Logging Properties
|
|
# =========================================================================
|
|
|
|
@property
|
|
def LOG_LEVEL(self) -> str:
|
|
"""Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)."""
|
|
return os.environ.get('LOG_LEVEL', 'INFO').upper()
|
|
|
|
def get_log_level_int(self) -> int:
|
|
"""Get logging level as integer for logging module."""
|
|
levels = {
|
|
'DEBUG': logging.DEBUG,
|
|
'INFO': logging.INFO,
|
|
'WARNING': logging.WARNING,
|
|
'ERROR': logging.ERROR,
|
|
'CRITICAL': logging.CRITICAL,
|
|
}
|
|
return levels.get(self.LOG_LEVEL, logging.INFO)
|
|
|
|
|
|
# Singleton instance for easy import
|
|
config = Config()
|