from enum import StrEnum from functools import cache from typing import Final, Literal, Self import sentry_sdk from pydantic import model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from src.worker import NonEmptyStr APPLICATION_NAME: Final = "lambda-assets-spatial-audio-validation" class AssetUploadType(StrEnum): ATMOS = "atmos" class Environment(StrEnum): DEV = "dev" QA = "qa" PROD = "prod" class Settings(BaseSettings): model_config = SettingsConfigDict(frozen=True, case_sensitive=True) ENVIRONMENT: Environment LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" SENTRY_DSN: str | None = None SQS_QUEUE_URL: NonEmptyStr MESSAGE_PROCESSING_TIMEOUT_MINUTES: int HEARTBEAT_INTERVAL_SECONDS: float @model_validator(mode="after") def _require_sentry_dsn_outside_dev(self) -> Self: if self.ENVIRONMENT != Environment.DEV and not self.SENTRY_DSN: error_message = f"SENTRY_DSN is required in {self.ENVIRONMENT}" raise ValueError(error_message) return self @property def token_manager_environment(self) -> Environment: if self.ENVIRONMENT == Environment.DEV: return Environment.QA return self.ENVIRONMENT @cache def get_settings() -> Settings: settings = Settings() if settings.SENTRY_DSN: sentry_sdk.init(dsn=settings.SENTRY_DSN, environment=settings.ENVIRONMENT) return settings def presigned_url_ttl_seconds() -> int: # A presigned URL only needs to outlive the message-processing budget. return get_settings().MESSAGE_PROCESSING_TIMEOUT_MINUTES * 60