84 lines
2.5 KiB
Python
Executable File
84 lines
2.5 KiB
Python
Executable File
# ================================================================
|
|
# SHARED CONFIGURATION UTILITIES
|
|
# shared/config/utils.py
|
|
# ================================================================
|
|
|
|
"""
|
|
Configuration utilities and helpers
|
|
"""
|
|
|
|
from typing import Dict, Any, Type
|
|
from shared.config.base import BaseServiceSettings
|
|
|
|
# Service settings registry
|
|
SERVICE_SETTINGS: Dict[str, Type[BaseServiceSettings]] = {
|
|
"gateway": GatewaySettings,
|
|
"auth-service": AuthSettings,
|
|
"training-service": TrainingSettings,
|
|
"forecasting-service": ForecastingSettings,
|
|
"data-service": DataSettings,
|
|
"tenant-service": TenantSettings,
|
|
"notification-service": NotificationSettings,
|
|
}
|
|
|
|
def get_settings_for_service(service_name: str) -> BaseServiceSettings:
|
|
"""
|
|
Get settings instance for a specific service
|
|
|
|
Args:
|
|
service_name: Name of the service
|
|
|
|
Returns:
|
|
BaseServiceSettings: Configured settings instance
|
|
|
|
Raises:
|
|
ValueError: If service name is not recognized
|
|
"""
|
|
if service_name not in SERVICE_SETTINGS:
|
|
raise ValueError(f"Unknown service: {service_name}. Available: {list(SERVICE_SETTINGS.keys())}")
|
|
|
|
settings_class = SERVICE_SETTINGS[service_name]
|
|
return settings_class()
|
|
|
|
def validate_all_service_configs() -> Dict[str, Any]:
|
|
"""
|
|
Validate configuration for all services
|
|
|
|
Returns:
|
|
Dict: Validation results for each service
|
|
"""
|
|
results = {}
|
|
|
|
for service_name, settings_class in SERVICE_SETTINGS.items():
|
|
try:
|
|
settings = settings_class()
|
|
results[service_name] = {
|
|
"status": "valid",
|
|
"config": {
|
|
"app_name": settings.APP_NAME,
|
|
"version": settings.VERSION,
|
|
"environment": settings.ENVIRONMENT,
|
|
"database_configured": bool(settings.DATABASE_URL),
|
|
"redis_configured": bool(settings.REDIS_URL),
|
|
"debug_mode": settings.DEBUG,
|
|
}
|
|
}
|
|
except Exception as e:
|
|
results[service_name] = {
|
|
"status": "error",
|
|
"error": str(e)
|
|
}
|
|
|
|
return results
|
|
|
|
def get_service_urls() -> Dict[str, str]:
|
|
"""
|
|
Get all service URLs from any service configuration
|
|
|
|
Returns:
|
|
Dict: Service name to URL mapping
|
|
"""
|
|
# Use auth service settings as reference (all services have same URLs)
|
|
settings = AuthSettings()
|
|
return settings.SERVICE_REGISTRY
|