diff --git a/src/python/role_play/chat/chat_logger.py b/src/python/role_play/chat/chat_logger.py index f015342..074fc3d 100644 --- a/src/python/role_play/chat/chat_logger.py +++ b/src/python/role_play/chat/chat_logger.py @@ -535,8 +535,10 @@ async def log_pcm_audio( """ # Security: Defensive check to ensure no PCM logging in production # even if environment checks are bypassed elsewhere - env = os.environ.get("ENV", "dev").lower() - if env == "prod" or env == "production": + from role_play.common.environment import resolve_environment + from role_play.common.models import Environment + env = resolve_environment() + if env == Environment.PROD: logger.debug("PCM audio logging disabled in production environment") return diff --git a/src/python/role_play/common/environment.py b/src/python/role_play/common/environment.py new file mode 100644 index 0000000..cc91baa --- /dev/null +++ b/src/python/role_play/common/environment.py @@ -0,0 +1,96 @@ +"""Unified environment parsing and resolution utilities. + +This module centralizes how the application determines its deployment +environment to avoid duplication and ambiguity across the codebase. + +Canonical environments: dev | beta | prod + +Accepted synonyms (case-insensitive): +- development -> dev +- production -> prod + +Priority for resolution (highest to lowest): +1) CONFIG_FILE (extracts env from filename like config/prod.yaml) +2) ENV +3) ENVIRONMENT +4) default: dev +""" + +from __future__ import annotations + +import os +import re +from typing import Optional + +from .models import Environment, EnvironmentInfo + + +def parse_environment_str(value: Optional[str]) -> Environment: + """Parse a string into the canonical Environment enum. + + Accepts canonical values (dev|beta|prod) and common synonyms + like development|production. + """ + if not value: + return Environment.DEV + + s = value.strip().lower() + # Map synonyms + if s == "development": + s = "dev" + elif s == "production": + s = "prod" + + try: + return Environment(s) + except ValueError: + # Default safely to DEV and let callers decide if they want to warn + return Environment.DEV + + +def resolve_environment(config_file: Optional[str] = None) -> Environment: + """Resolve the current environment using a consistent precedence order. + + Precedence: + 1) CONFIG_FILE (env extracted from its basename, e.g., beta.yaml) + 2) ENV + 3) ENVIRONMENT + 4) default to dev + """ + # 1) CONFIG_FILE (argument overrides env var for testability) + config_file = config_file or os.getenv("CONFIG_FILE") + if config_file: + # Try to extract the environment from the filename + # e.g., /app/config/beta.yaml -> beta + m = re.search(r"/(\\w+)\\.yaml$", config_file) + if m: + return parse_environment_str(m.group(1)) + + # 2) ENV + env = os.getenv("ENV") + if env: + return parse_environment_str(env) + + # 3) ENVIRONMENT (supports development/production synonyms) + env2 = os.getenv("ENVIRONMENT") + if env2: + return parse_environment_str(env2) + + # 4) fallback + return Environment.DEV + + +def get_environment_info() -> EnvironmentInfo: + """Return an EnvironmentInfo snapshot using the unified resolver.""" + env = resolve_environment() + return EnvironmentInfo( + name=env, + is_production=(env == Environment.PROD), + is_development=(env == Environment.DEV), + ) + + +def environment_name() -> str: + """Convenience accessor for the current environment name as a string.""" + return resolve_environment().value + diff --git a/src/python/role_play/common/logging_config.py b/src/python/role_play/common/logging_config.py index eab6713..10832bc 100644 --- a/src/python/role_play/common/logging_config.py +++ b/src/python/role_play/common/logging_config.py @@ -33,7 +33,11 @@ def format(self, record: logging.LogRecord) -> str: log_entry.update(record.extra_fields) # Add environment info - log_entry["environment"] = os.getenv("ENV", "unknown") + try: + from .environment import environment_name + log_entry["environment"] = environment_name() + except Exception: + log_entry["environment"] = os.getenv("ENV", "unknown") log_entry["service"] = os.getenv("SERVICE_NAME", "rps") log_entry["version"] = os.getenv("GIT_VERSION", "unknown") @@ -63,7 +67,13 @@ def setup_logging(log_level: str = "INFO", use_structured: bool = True) -> None: console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(numeric_level) - if use_structured and os.getenv("ENV", "dev") != "dev": + try: + from .environment import resolve_environment + is_dev = (resolve_environment().value == "dev") + except Exception: + is_dev = (os.getenv("ENV", "dev") == "dev") + + if use_structured and not is_dev: # Use structured JSON logging for non-dev environments formatter = StructuredFormatter() else: @@ -83,13 +93,19 @@ def setup_logging(log_level: str = "INFO", use_structured: bool = True) -> None: # Log the configuration logger = logging.getLogger(__name__) + try: + from .environment import environment_name as _env_name + env_name_value = _env_name() + except Exception: + env_name_value = os.getenv("ENV", "dev") + logger.info( "Logging configured", extra={ "extra_fields": { "log_level": log_level, "use_structured": use_structured, - "environment": os.getenv("ENV", "dev") + "environment": env_name_value, } } ) @@ -105,4 +121,4 @@ def get_logger(name: str) -> logging.Logger: Returns: Logger instance """ - return logging.getLogger(name) \ No newline at end of file + return logging.getLogger(name) diff --git a/src/python/role_play/common/storage_factory.py b/src/python/role_play/common/storage_factory.py index 6b329de..0a32ebc 100644 --- a/src/python/role_play/common/storage_factory.py +++ b/src/python/role_play/common/storage_factory.py @@ -9,6 +9,7 @@ FileStorageConfig, GCSStorageConfig, S3StorageConfig ) from .exceptions import StorageError +from .environment import resolve_environment def create_storage_backend( @@ -111,11 +112,7 @@ def create_storage_from_env(environment: Union[Environment, str] = None) -> Stor """ # Auto-detect environment if not provided if environment is None: - env_str = os.getenv("ENV", "dev").lower() - try: - environment = Environment(env_str) - except ValueError: - raise StorageError(f"Invalid ENV environment variable: {env_str}") + environment = resolve_environment() # Get storage type storage_type = os.getenv("STORAGE_TYPE") @@ -279,4 +276,4 @@ def validate_storage_config(config: StorageConfigUnion) -> None: retry_attempts: 3 retry_delay_seconds: 1.0 """ -} \ No newline at end of file +} diff --git a/src/python/role_play/server/base_server.py b/src/python/role_play/server/base_server.py index c0b480f..1282bd6 100644 --- a/src/python/role_play/server/base_server.py +++ b/src/python/role_play/server/base_server.py @@ -71,9 +71,10 @@ async def lifespan(app: FastAPI): @self.app.get("/health", tags=["Health"], include_in_schema=False) async def health_check(): """Health check endpoint for monitoring and Cloud Run.""" + from role_play.common.environment import environment_name return { "status": "healthy", - "environment": os.getenv("ENV", "unknown"), + "environment": environment_name(), "version": os.getenv("GIT_VERSION", "unknown"), "service": os.getenv("SERVICE_NAME", "rps") } @@ -156,7 +157,9 @@ async def serve_spa_index(full_path: str): return FileResponse(index_html_path) else: # In development, the frontend might not be built yet - if os.getenv("ENV", "dev") == "dev": + from role_play.common.environment import resolve_environment + from role_play.common.models import Environment + if resolve_environment() == Environment.DEV: return { "message": "Frontend not found. In development, use 'npm run dev' for the frontend.", "static_dir": static_files_dir @@ -167,4 +170,4 @@ async def serve_spa_index(full_path: str): def get_app(self) -> FastAPI: """Return the FastAPI application instance.""" - return self.app \ No newline at end of file + return self.app diff --git a/src/python/role_play/server/config.py b/src/python/role_play/server/config.py index f44a6f3..f2b7aa9 100644 --- a/src/python/role_play/server/config.py +++ b/src/python/role_play/server/config.py @@ -109,18 +109,12 @@ def __init__(self, **data): def get_config(environment: Optional[str] = None) -> ServerConfig: """ - Get configuration based on environment. + Deprecated: Use role_play.server.config_loader.get_config instead. - Args: - environment: Environment name (development, production) or None for auto-detection - - Returns: - ServerConfig: Configuration instance + This wrapper delegates to the unified config loader to avoid confusion + between different environment names and selection logic. + Accepts canonical envs (dev|beta|prod) and common synonyms. """ - if environment is None: - environment = os.getenv("ENVIRONMENT", "development") - - if environment == "production": - return ProductionConfig() - else: - return DevelopmentConfig() + # Import here to avoid circular import at module load time + from .config_loader import get_config as _get_config + return _get_config(environment) diff --git a/src/python/role_play/server/config_loader.py b/src/python/role_play/server/config_loader.py index beefd4e..c1055ca 100644 --- a/src/python/role_play/server/config_loader.py +++ b/src/python/role_play/server/config_loader.py @@ -10,19 +10,13 @@ import os import yaml -from enum import Enum from pathlib import Path from typing import Dict, Any, Optional from dotenv import load_dotenv from .config import ServerConfig, DevelopmentConfig, ProductionConfig - - -class Environment(Enum): - """Supported environments.""" - DEV = "dev" - BETA = "beta" - PROD = "prod" +from ..common.models import Environment +from ..common.environment import resolve_environment class ConfigLoader: @@ -190,26 +184,12 @@ def get_config(self, environment: Optional[str] = None, force_reload: bool = Fal # Load environment variables first self.load_environment_variables() - # Determine and validate environment + # Determine and validate environment using unified resolver if environment is None: - # Check CONFIG_FILE first, then ENV, then ENVIRONMENT, then default to dev - config_file = os.getenv("CONFIG_FILE") - if config_file: - # Extract environment from config file path (e.g., /app/config/beta.yaml -> beta) - import re - match = re.search(r'/(\w+)\.yaml$', config_file) - if match: - environment = match.group(1) - else: - environment = "dev" - else: - environment = os.getenv("ENV", os.getenv("ENVIRONMENT", "dev")) - - try: + env_enum = resolve_environment() + else: env_enum = Environment(environment) - except ValueError: - supported_envs = [e.value for e in Environment] - raise ValueError(f"Unsupported environment '{environment}'. Supported environments: {supported_envs}") + # Load YAML config yaml_config = self.load_yaml_config(env_enum) @@ -253,4 +233,4 @@ def get_config(environment: Optional[str] = None) -> ServerConfig: def reset_config(): """Reset the global config loader (for testing).""" global _config_loader - _config_loader = None \ No newline at end of file + _config_loader = None diff --git a/src/python/role_play/server/dependencies.py b/src/python/role_play/server/dependencies.py index 1bedb1b..3a0b121 100644 --- a/src/python/role_play/server/dependencies.py +++ b/src/python/role_play/server/dependencies.py @@ -12,6 +12,7 @@ from ..common.storage import StorageBackend, FileStorage, FileStorageConfig, LockConfig from ..common.storage_factory import create_storage_backend from ..common.models import User, UserRole, Environment, EnvironmentInfo +from ..common.environment import resolve_environment, get_environment_info as resolved_env_info from ..common.exceptions import AuthenticationError, TokenExpiredError from .config_loader import get_config, ServerConfig from ..chat.chat_logger import ChatLogger @@ -24,21 +25,7 @@ @lru_cache(maxsize=None) def get_environment_info() -> EnvironmentInfo: """Provides detailed information about the current deployment environment.""" - env_str = os.getenv("ENV", "dev") - try: - env_enum = Environment(env_str) - except ValueError: - logger.warning(f"Unknown environment '{env_str}', defaulting to DEV") - env_enum = Environment.DEV - - is_prod = (env_enum == Environment.PROD) - is_dev = (env_enum == Environment.DEV) - - return EnvironmentInfo( - name=env_enum, - is_production=is_prod, - is_development=is_dev - ) + return resolved_env_info() @lru_cache(maxsize=None) @@ -55,19 +42,13 @@ def get_storage_backend() -> StorageBackend: """ config = get_server_config() - # Determine environment - environment = os.getenv("ENVIRONMENT", "dev") - try: - env_enum = Environment(environment) - except ValueError: - # Default to dev for unknown environments - env_enum = Environment.DEV - logger.warning(f"Unknown environment '{environment}', defaulting to DEV") + # Determine environment via unified resolver + env_enum = resolve_environment() # Use storage configuration if config.storage: backend = create_storage_backend(config.storage, env_enum) - logger.info(f"Storage backend: {type(backend).__name__} for {environment}") + logger.info(f"Storage backend: {type(backend).__name__} for {env_enum.value}") return backend else: raise ValueError("Storage configuration is required")