refactor env proposal

src/summary/summary/api/route/tasks.py

@@ -12,8 +12,6 @@ from summary.core.celery_worker import (
 )
 from summary.core.config import get_settings
 
-settings = get_settings()
-
 
 class TranscribeSummarizeTaskCreation(BaseModel):
     """Transcription and summarization parameters."""
@@ -34,10 +32,11 @@ class TranscribeSummarizeTaskCreation(BaseModel):
     @classmethod
     def validate_language(cls, v):
         """Validate 'language' parameter."""
-        if v is not None and v not in settings.whisperx_allowed_languages:
+        allowed = get_settings().whisperx_allowed_languages
+        if v is not None and v not in allowed:
             raise ValueError(
                 f"Language '{v}' is not allowed. "
-                f"Allowed languages: {', '.join(settings.whisperx_allowed_languages)}"
+                f"Allowed languages: {', '.join(allowed)}"
             )
         return v
 
@@ -62,7 +61,7 @@ async def create_transcribe_summarize_task(request: TranscribeSummarizeTaskCreat
             request.download_link,
             request.context_language,
         ],
-        queue=settings.transcribe_queue,
+        queue=get_settings().transcribe_queue,
     )
 
     return {"id": task.id, "message": "Task created"}

src/summary/summary/core/analytics.py

@@ -12,7 +12,6 @@ from posthog import Posthog
 from summary.core.config import get_settings
 
 logger = get_task_logger(__name__)
-settings = get_settings()
 
 
 class AnalyticsException(Exception):
@@ -26,6 +25,7 @@ class Analytics:
 
     def __init__(self):
         """Initialize a client if settings are configure."""
+        settings = get_settings()
         self._client = None
         if settings.posthog_api_key and settings.posthog_enabled:
             logger.info("Initialize analytics client")
@@ -71,6 +71,7 @@ class MetadataManager:
 
     def __init__(self):
         """Initialize the task tracker with analytics client."""
+        settings = get_settings()
         self._redis = redis.from_url(settings.task_tracker_redis_url)
         self._key_prefix = settings.task_tracker_prefix
         self._analytics = get_analytics()
@@ -117,7 +118,7 @@ class MetadataManager:
         start_time = time.time()
         initial_metadata = {
             "start_time": start_time,
-            "asr_model": settings.whisperx_asr_model,
+            "asr_model": get_settings().whisperx_asr_model,
             "retries": 0,
             "filename": filename,
             "email": email,

src/summary/summary/core/config.py

@@ -1,5 +1,6 @@
 """Application configuration and settings."""
 
+import os
 from functools import lru_cache
 from typing import Annotated, List, Literal, Optional, Set
 
@@ -91,9 +92,37 @@ class Settings(BaseSettings):
     task_tracker_prefix: str = "task_metadata:"
 
 
+class TestSettings(Settings):
+    """Settings with safe defaults for testing."""
+
+    model_config = SettingsConfigDict(env_file=None)
+
+    app_api_token: SecretStr = SecretStr("test-api-token")
+    aws_storage_bucket_name: str = "test-bucket"
+    aws_s3_endpoint_url: str = "http://localhost:9000"
+    aws_s3_access_key_id: str = "test-access-key"
+    aws_s3_secret_access_key: SecretStr = SecretStr("test-secret-key")
+    aws_s3_secure_access: bool = False
+    whisperx_api_key: SecretStr = SecretStr("test-whisperx-key")
+    whisperx_base_url: str = "http://localhost:8000/v1"
+    llm_base_url: str = "http://localhost:8001/v1"
+    llm_api_key: SecretStr = SecretStr("test-llm-key")
+    llm_model: str = "test-model"
+    webhook_api_token: SecretStr = SecretStr("test-webhook-token")
+    webhook_url: str = "http://localhost:8002/webhook"
+    celery_broker_url: str = "memory://"
+    celery_result_backend: str = "cache+memory://"
+    posthog_enabled: bool = False
+    sentry_is_enabled: bool = False
+    langfuse_enabled: bool = False
+    task_tracker_redis_url: str = "redis://localhost:6379/0"
+
+
 @lru_cache
 def get_settings():
     """Load and cache application settings."""
+    if os.environ.get("SUMMARY_ENV") == "test":
+        return TestSettings()
     return Settings()
 
 

src/summary/summary/core/file_service.py

@@ -13,9 +13,6 @@ from minio.error import MinioException, S3Error
 
 from summary.core.config import get_settings
 
-settings = get_settings()
-
-
 logger = logging.getLogger(__name__)
 
 
@@ -31,23 +28,24 @@ class FileService:
     def __init__(self):
         """Initialize FileService with MinIO client and configuration."""
         endpoint = (
-            settings.aws_s3_endpoint_url.removeprefix("https://")
+            get_settings()
+            .aws_s3_endpoint_url.removeprefix("https://")
             .removeprefix("http://")
             .rstrip("/")
         )
 
         self._minio_client = Minio(
             endpoint,
-            access_key=settings.aws_s3_access_key_id,
-            secret_key=settings.aws_s3_secret_access_key.get_secret_value(),
-            secure=settings.aws_s3_secure_access,
+            access_key=get_settings().aws_s3_access_key_id,
+            secret_key=get_settings().aws_s3_secret_access_key.get_secret_value(),
+            secure=get_settings().aws_s3_secure_access,
         )
 
-        self._bucket_name = settings.aws_storage_bucket_name
+        self._bucket_name = get_settings().aws_storage_bucket_name
         self._stream_chunk_size = 32 * 1024
 
-        self._allowed_extensions = settings.recording_allowed_extensions
-        self._max_duration = settings.recording_max_duration
+        self._allowed_extensions = get_settings().recording_allowed_extensions
+        self._max_duration = get_settings().recording_max_duration
 
     def _download_from_minio(self, remote_object_key) -> Path:
         """Download file from MinIO to local temporary file.
@@ -174,7 +172,7 @@ class FileService:
 
             extension = downloaded_path.suffix.lower()
 
-            if extension in settings.recording_video_extensions:
+            if extension in get_settings().recording_video_extensions:
                 logger.info("Video file detected, extracting audio...")
                 extracted_audio_path = self._extract_audio_from_video(downloaded_path)
                 processed_path = extracted_audio_path

src/summary/summary/core/llm_service.py

@@ -8,9 +8,6 @@ from langfuse import Langfuse
 
 from summary.core.config import get_settings
 
-settings = get_settings()
-
-
 logger = logging.getLogger(__name__)
 
 
@@ -34,28 +31,28 @@ class LLMObservability:
         self.session_id = session_id
         self.user_id = user_id
 
-        if settings.langfuse_enabled:
+        if get_settings().langfuse_enabled:
 
             def masking_function(data, **kwargs):
                 if (
                     user_has_tracing_consent
-                    or settings.langfuse_environment != "production"
+                    or get_settings().langfuse_environment != "production"
                 ):
                     return data
 
                 return "[REDACTED]"
 
-            if not settings.langfuse_secret_key:
+            if not get_settings().langfuse_secret_key:
                 raise ValueError(
                     "langfuse_secret_key is not configured. "
                     "Please set the secret key or disable Langfuse."
                 )
 
             self._observability_client = Langfuse(
-                secret_key=settings.langfuse_secret_key.get_secret_value(),
-                public_key=settings.langfuse_public_key,
-                host=settings.langfuse_host,
-                environment=settings.langfuse_environment,
+                secret_key=get_settings().langfuse_secret_key.get_secret_value(),
+                public_key=get_settings().langfuse_public_key,
+                host=get_settings().langfuse_host,
+                environment=get_settings().langfuse_environment,
                 mask=masking_function,
             )
 
@@ -72,8 +69,8 @@ class LLMObservability:
         to Langfuse for observability when enabled.
         """
         base_args = {
-            "base_url": settings.llm_base_url,
-            "api_key": settings.llm_api_key.get_secret_value(),
+            "base_url": get_settings().llm_base_url,
+            "api_key": get_settings().llm_api_key.get_secret_value(),
         }
 
         if not self.is_enabled:
@@ -120,7 +117,7 @@ class LLMService:
         """
         try:
             params: dict[str, Any] = {
-                "model": settings.llm_model,
+                "model": get_settings().llm_model,
                 "messages": [
                     {"role": "system", "content": system_prompt},
                     {"role": "user", "content": user_prompt},

src/summary/summary/core/transcript_formatter.py

@@ -6,8 +6,6 @@ from typing import Optional, Tuple
 from summary.core.config import get_settings
 from summary.core.locales import LocaleStrings
 
-settings = get_settings()
-
 logger = logging.getLogger(__name__)
 
 
@@ -23,7 +21,7 @@ class TranscriptFormatter:
 
     def __init__(self, locale: LocaleStrings):
         """Initialize formatter with settings and locale."""
-        self.hallucination_patterns = settings.hallucination_patterns
+        self.hallucination_patterns = get_settings().hallucination_patterns
         self._locale = locale
 
     def _get_segments(self, transcription):

src/summary/summary/core/webhook_service.py

@@ -9,8 +9,6 @@ from urllib3.util import Retry
 
 from summary.core.config import get_settings
 
-settings = get_settings()
-
 logger = logging.getLogger(__name__)
 
 
@@ -18,9 +16,9 @@ def _create_retry_session():
     """Create an HTTP session configured with retry logic."""
     session = Session()
     retries = Retry(
-        total=settings.webhook_max_retries,
-        backoff_factor=settings.webhook_backoff_factor,
-        status_forcelist=settings.webhook_status_forcelist,
+        total=get_settings().webhook_max_retries,
+        backoff_factor=get_settings().webhook_backoff_factor,
+        status_forcelist=get_settings().webhook_status_forcelist,
         allowed_methods={"POST"},
     )
     session.mount("https://", HTTPAdapter(max_retries=retries))
@@ -31,7 +29,9 @@ def _post_with_retries(url, data):
     """Send POST request with automatic retries."""
     session = _create_retry_session()
     session.headers.update(
-        {"Authorization": f"Bearer {settings.webhook_api_token.get_secret_value()}"}
+        {
+            "Authorization": f"Bearer {get_settings().webhook_api_token.get_secret_value()}"
+        }
     )
     try:
         response = session.post(url, json=data)
@@ -53,10 +53,10 @@ def submit_content(content, title, email, sub):
         "sub": sub,
     }
 
-    logger.debug("Submitting to %s", settings.webhook_url)
+    logger.debug("Submitting to %s", get_settings().webhook_url)
     logger.debug("Request payload: %s", json.dumps(data, indent=2))
 
-    response = _post_with_retries(settings.webhook_url, data)
+    response = _post_with_retries(get_settings().webhook_url, data)
 
     try:
         response_data = response.json()

src/summary/tests/conftest.py

@@ -2,25 +2,6 @@
 
 import os
 
-# Set required environment variables before any summary module imports.
-# This is necessary because several modules call get_settings() at module level,
-# which validates env vars via Pydantic Settings.
-os.environ.setdefault("APP_API_TOKEN", "test-api-token")
-os.environ.setdefault("AWS_STORAGE_BUCKET_NAME", "test-bucket")
-os.environ.setdefault("AWS_S3_ENDPOINT_URL", "http://localhost:9000")
-os.environ.setdefault("AWS_S3_ACCESS_KEY_ID", "test-access-key")
-os.environ.setdefault("AWS_S3_SECRET_ACCESS_KEY", "test-secret-key")
-os.environ.setdefault("AWS_S3_SECURE_ACCESS", "false")
-os.environ.setdefault("WHISPERX_API_KEY", "test-whisperx-key")
-os.environ.setdefault("WHISPERX_BASE_URL", "http://localhost:8000/v1")
-os.environ.setdefault("LLM_BASE_URL", "http://localhost:8001/v1")
-os.environ.setdefault("LLM_API_KEY", "test-llm-key")
-os.environ.setdefault("LLM_MODEL", "test-model")
-os.environ.setdefault("WEBHOOK_API_TOKEN", "test-webhook-token")
-os.environ.setdefault("WEBHOOK_URL", "http://localhost:8002/webhook")
-os.environ.setdefault("CELERY_BROKER_URL", "memory://")
-os.environ.setdefault("CELERY_RESULT_BACKEND", "cache+memory://")
-os.environ.setdefault("POSTHOG_ENABLED", "false")
-os.environ.setdefault("SENTRY_IS_ENABLED", "false")
-os.environ.setdefault("LANGFUSE_ENABLED", "false")
-os.environ.setdefault("TASK_TRACKER_REDIS_URL", "redis://localhost:6379/0")
+# Activate TestSettings (safe defaults for all required env vars)
+# before any summary module is imported.
+os.environ["SUMMARY_ENV"] = "test"

src/summary/tests/integration/test_api_tasks.py

@@ -9,11 +9,11 @@ import responses
 
 from summary.core.config import get_settings
 
-settings = get_settings()
-
 API_PREFIX = "/api/v1"
-AUTH_HEADER = {"Authorization": f"Bearer {settings.app_api_token.get_secret_value()}"}
-WEBHOOK_URL = settings.webhook_url
+AUTH_HEADER = {
+    "Authorization": f"Bearer {get_settings().app_api_token.get_secret_value()}"
+}
+WEBHOOK_URL = get_settings().webhook_url
 
 
 class TestTranscribeSummarizeFlow:

src/summary/tests/unit/test_celery_worker.py

@@ -16,9 +16,7 @@ from summary.core.celery_worker import (
 from summary.core.config import get_settings
 from summary.core.file_service import FileServiceException
 
-settings = get_settings()
-
-WEBHOOK_URL = settings.webhook_url
+WEBHOOK_URL = get_settings().webhook_url
 
 # ---------------------------------------------------------------------------
 # transcribe_audio
@@ -241,7 +239,7 @@ class TestSummarizeTranscription:
         # Verify auth header was sent
         assert (
             webhook_request.request.headers["Authorization"]
-            == f"Bearer {settings.webhook_api_token.get_secret_value()}"
+            == f"Bearer {get_settings().webhook_api_token.get_secret_value()}"
         )
 
         # LLM was called for: tldr, plan, part A, part B, next-steps, cleaning