test_workflow

workspace/sprint_1_2/CODEBASE/backend/implementation/config.py

@@ -1,47 +1,131 @@
+from __future__ import annotations
+
+import json
 import os
-from pathlib import Path
-from typing import Dict
+from typing import Dict, List, Optional, Tuple
 
-SECRETS_DIR = Path(__file__).resolve().parent.parent.parent.parent.parent.parent / "secrets"
+from pydantic import Field, HttpUrl, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
-def _load_secret(name: str, filename: str) -> str:
-    file_path = SECRETS_DIR / filename
-    env_file = os.getenv(f"{name}_FILE")
-    if env_file:
-        resolved = Path(env_file)
-        if resolved.exists():
-            with open(resolved, "r", encoding="utf-8") as f:
-                return f.read().strip()
-    if file_path.exists():
-        with open(file_path, "r", encoding="utf-8") as f:
-            return f.read().strip()
-    raise RuntimeError(
-        f"Required secret {name} not found at {file_path} or via {name}_FILE env var"
+
+def _require_env(name: str) -> str:
+    """Require a secret from environment variable only.
+    
+    In production/Gitea Actions, this comes from repository secrets.
+    No file fallback to avoid accidental secret leakage into the repo.
+    """
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(
+            f"Required secret {name} not found. Set {name} environment variable. "
+            f"In Gitea Actions, add it as a repository secret."
+        )
+    return value
+
+
+_CORS_ORIGINS_DEFAULT = ",".join(
+    [
+        "http://localhost:3000",
+        "http://localhost:5173",
+        "http://localhost:4173",
+        "http://127.0.0.1:5173",
+    ]
+)
+
+
+def _parse_cors_origins(value: Optional[str]) -> List[str]:
+    if not value:
+        return [o.strip() for o in _CORS_ORIGINS_DEFAULT.split(",") if o.strip()]
+    try:
+        parsed = json.loads(value)
+        if isinstance(parsed, list):
+            return [str(item).strip() for item in parsed if str(item).strip()]
+    except (json.JSONDecodeError, TypeError):
+        pass
+    return [origin.strip() for origin in value.split(",") if origin.strip()]
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
     )
 
-# Endpoints (environment-provided, no hardcoded fallback for production)
+    # Triton
+    triton_endpoint: Optional[HttpUrl] = Field(default=None, validation_alias="TRITON_ENDPOINT")
+
+    # Server
+    cv_inference_host: str = Field(default="127.0.0.1", validation_alias="CV_INFERENCE_HOST")
+    cv_inference_port: int = Field(
+        default=8001, ge=1, le=65535, validation_alias="CV_INFERENCE_PORT"
+    )
+
+    # CORS - keep raw string to avoid pydantic-settings JSON-list parsing pitfalls
+    cors_origins_raw: str = Field(
+        default=_CORS_ORIGINS_DEFAULT,
+        validation_alias="CORS_ORIGINS",
+    )
+
+    @property
+    def cors_origins(self) -> List[str]:
+        return _parse_cors_origins(self.cors_origins_raw)
+
+    # Domain
+    # external_host: Optional[HttpUrl] = Field(default=None, validation_alias="EXTERNAL_HOST") # currently deprecated for no use
+    base_url: Optional[HttpUrl] = Field(default=None, validation_alias="BASE_URL") # can use for routing toward other API later
+
+    # Other settings
+    project_id: str = Field(default="vkist-project", validation_alias="VERTEX_AI_PROJECT")
+    location: str = Field(default="asia-southeast1", validation_alias="VERTEX_AI_LOCATION")
+    temp_dir: str = Field(default="/tmp/analysis_jobs", validation_alias="TEMP_DIR")
+    vertex_ai_model: str = Field(default="medgemma", validation_alias="VERTEX_AI_MODEL")
+    redis_host: str = Field(default="localhost", validation_alias="REDIS_HOST")
+    redis_port: int = Field(default=6379, validation_alias="REDIS_PORT")
+    redis_db: int = Field(default=0, validation_alias="REDIS_DB")
+    clahe_clip_limit: float = Field(default=2.0, validation_alias="CLAHE_CLIP_LIMIT")
+    clahe_tile_size: Tuple[int, int] = Field(default=(8, 8), validation_alias="CLAHE_TILE_SIZE")
+
+    @field_validator("clahe_tile_size", mode="before")
+    @classmethod
+    def parse_tile_size(cls, v):
+        if isinstance(v, str):
+            parts = v.split(",")
+            if len(parts) == 2:
+                return (int(parts[0].strip()), int(parts[1].strip()))
+        return v
+
+
+settings = Settings()
+
+# Endpoints
 MODAL_MEDGEMMA_ENDPOINT = os.getenv("MODAL_MEDGEMMA_ENDPOINT")
 VERTEX_AI_GEMINI_ENDPOINT = os.getenv("VERTEX_AI_GEMINI_ENDPOINT")
 
-# Secrets (must be present in PILOT_PROJECT/secrets or env)
-GCP_ACCESS_TOKEN = _load_secret("GCP_ACCESS_TOKEN", "gcp_access_token.txt")
-MEDGEMMA_API_KEY = _load_secret("MEDGEMMA_API_KEY", "modal_api_key.txt")
+# Secrets - must come from environment variables only
+# In Gitea Actions, these are set via repository secrets
+# In local development, set via .env or shell environment
+GCP_ACCESS_TOKEN = os.getenv("GCP_ACCESS_TOKEN")
+MEDGEMMA_API_KEY = os.getenv("MEDGEMMA_API_KEY")
 
-PROJECT_ID = os.getenv("VERTEX_AI_PROJECT", "vkist-project")
-LOCATION = os.getenv("VERTEX_AI_LOCATION", "asia-southeast1")
+# Legacy module-level constants for backward compatibility.
+# These now derive from the validated settings model instead of raw os.getenv().
+PROJECT_ID = settings.project_id
+LOCATION = settings.location
 
-TRITON_ENDPOINT = os.getenv("TRITON_ENDPOINT", "http://localhost:8000")
-TEMP_DIR = os.getenv("TEMP_DIR", "/tmp/analysis_jobs")
+TRITON_ENDPOINT = (
+    str(settings.triton_endpoint).rstrip("/") if settings.triton_endpoint else None
+)
 
-# LLM Configuration
-VERTEX_AI_PROJECT = os.getenv("VERTEX_AI_PROJECT", "vkist-project")
-VERTEX_AI_LOCATION = os.getenv("VERTEX_AI_LOCATION", "asia-southeast1")
-VERTEX_AI_MODEL = os.getenv("VERTEX_AI_MODEL", "medgemma")
+TEMP_DIR = settings.temp_dir
 
-# Redis Configuration
-REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
-REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
-REDIS_DB = int(os.getenv("REDIS_DB", "0"))
+VERTEX_AI_PROJECT = settings.project_id
+VERTEX_AI_LOCATION = settings.location
+VERTEX_AI_MODEL = settings.vertex_ai_model
+
+REDIS_HOST = settings.redis_host
+REDIS_PORT = settings.redis_port
+REDIS_DB = settings.redis_db
 
 DEFAULT_MODEL_VERSIONS = {
     "angle": "angle_classify_convnext_tiny",
@@ -50,8 +134,8 @@ DEFAULT_MODEL_VERSIONS = {
     "segmentation_post": "segmentation_model_post_deeplabv3_resnet101",
 }
 
-CLAHE_CLIP_LIMIT = float(os.getenv("CLAHE_CLIP_LIMIT", "2.0"))
-CLAHE_TILE_SIZE = tuple(int(x) for x in os.getenv("CLAHE_TILE_SIZE", "8,8").split(","))
+CLAHE_CLIP_LIMIT = settings.clahe_clip_limit
+CLAHE_TILE_SIZE = settings.clahe_tile_size
 
 
 def get_model_name(task: str, model_versions: Dict[str, str] | None = None) -> str:
@@ -72,4 +156,3 @@ def get_segmentation_model(angle_class: str, model_versions: Dict[str, str] | No
     angle_type = get_angle_type(angle_class)
     task = "segmentation_sup" if angle_type == "sup" else "segmentation_post"
     return get_model_name(task, model_versions)
-