File size: 7,056 Bytes
e391a84 77f2d58 e391a84 77f2d58 e391a84 77f2d58 e391a84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """
shared/config.py
────────────────
Application configuration loaded from environment variables.
Uses Pydantic Settings so every variable is validated + type-safe.
Supabase support:
• DATABASE_URL should point to Supabase PostgreSQL (port 5432 direct, or
port 6543 for the Connection Pooler in Transaction mode).
• SUPABASE_URL / SUPABASE_ANON_KEY are optional — only needed if you use
the Supabase Python client SDK for storage, auth, or realtime features.
"""
from __future__ import annotations
from functools import lru_cache
import os
# Detect Kaggle environment and auto-load secrets into environment variables
if "KAGGLE_KERNEL_RUN_TYPE" in os.environ or os.path.exists("/kaggle"):
try:
from kaggle_secrets import UserSecretsClient
user_secrets = UserSecretsClient()
for key in ["DATABASE_URL", "RABBITMQ_URL", "USE_MOCK_MODEL"]:
try:
val = user_secrets.get_secret(key)
if val:
os.environ[key] = val
except Exception:
pass
except ImportError:
pass
from pathlib import Path
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
# Resolve absolute path to the .env file in the project root
_ROOT_DIR = Path(__file__).resolve().parent.parent.parent
_ENV_FILE = _ROOT_DIR / ".env"
class Settings(BaseSettings):
"""
Central configuration object.
Priority order (highest → lowest):
1. OS environment variables
2. .env file in the project root
3. Default values defined here
"""
model_config = SettingsConfigDict(
env_file=str(_ENV_FILE),
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# ── Database ──────────────────────────────────────────────────────────────
database_url: str = Field(
default="sqlite+aiosqlite:///./bp_monitoring.db",
description=(
"Async database connection string. "
"• postgresql+asyncpg://postgres.[ref]:[pw]@[host]:5432/postgres (Supabase direct) "
"• postgresql+asyncpg://postgres.[ref]:[pw]@[host]:6543/postgres (Supabase pooler) "
"• sqlite+aiosqlite:///./bp_monitoring.db (local dev)"
),
)
# Connection pool tuning (ignored for SQLite)
db_pool_size: int = Field(
default=5,
description=(
"SQLAlchemy connection pool size. "
"Supabase free tier allows up to 60 connections; keep this ≤ 10."
),
)
db_max_overflow: int = Field(
default=10,
description="Extra connections allowed above pool_size during peak load.",
)
db_pool_recycle: int = Field(
default=1800,
description="Recycle idle connections after N seconds (30 min default).",
)
# ── Supabase (optional — for SDK features beyond raw SQL) ─────────────────
supabase_url: str = Field(
default="",
description="Supabase project URL (https://[project-ref].supabase.co). Optional.",
)
supabase_anon_key: str = Field(
default="",
description="Supabase anonymous/public API key. Optional.",
)
# ── Message Broker ────────────────────────────────────────────────────────
rabbitmq_url: str = Field(
default="amqp://guest:guest@localhost:5672/",
description="RabbitMQ connection URL. Use amqps:// for CloudAMQP (SSL).",
)
# ── FastAPI Server ────────────────────────────────────────────────────────
app_host: str = Field(default="0.0.0.0", description="Bind host.")
app_port: int = Field(default=7860, description="Bind port (7860 for HF Spaces).")
# ── Application ───────────────────────────────────────────────────────────
debug: bool = Field(default=False, description="Enable debug mode.")
log_level: str = Field(default="INFO", description="Logging level.")
# ── AI Model ──────────────────────────────────────────────────────────────
gan_checkpoint_path: str = Field(
default="./models/gan_checkpoint.pt",
description="Path to GAN model checkpoint.",
)
vgtlnet_checkpoint_path: str = Field(
default="./models/vgtlnet_checkpoint.pt",
description="Path to VGTL-Net model checkpoint.",
)
use_mock_model: bool = Field(
default=True,
description=(
"Use MockModelService instead of real GAN+VGTL-Net. "
"Set to false only when checkpoints are available."
),
)
# ── Validators ────────────────────────────────────────────────────────────
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
valid = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
upper = v.upper()
if upper not in valid:
raise ValueError(f"log_level must be one of {valid}")
return upper
@field_validator("app_port")
@classmethod
def validate_port(cls, v: int) -> int:
if not (1 <= v <= 65535):
raise ValueError("app_port must be between 1 and 65535")
return v
@field_validator("db_pool_size")
@classmethod
def validate_pool_size(cls, v: int) -> int:
if v < 1:
raise ValueError("db_pool_size must be at least 1")
return v
# ── Computed Helpers ──────────────────────────────────────────────────────
@property
def is_supabase(self) -> bool:
"""True when DATABASE_URL points to Supabase (supabase.co host)."""
return "supabase.co" in self.database_url
@property
def is_sqlite(self) -> bool:
"""True when DATABASE_URL uses SQLite (local dev)."""
return self.database_url.startswith("sqlite")
@property
def uses_pooler(self) -> bool:
"""True when connecting via Supabase's pgBouncer pooler (port 6543)."""
return ":6543/" in self.database_url
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""
Return a cached singleton Settings instance.
"""
return Settings()
|