Spaces:
Runtime error
Runtime error
File size: 4,945 Bytes
31e8a9f | 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 | """Application settings loaded from environment variables / the repo-root `.env`.
Every setting is documented in `.env.example`. Settings are read once and
cached via :func:`get_settings`; import the module-level ``settings`` object for
convenience.
"""
from __future__ import annotations
from functools import lru_cache
from ipaddress import ip_network
from pathlib import Path
from typing import Literal
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
#: Repository root (``backend/app/config.py`` -> two parents up from ``app``).
REPO_ROOT = Path(__file__).resolve().parents[2]
def _normalize_database_url(url: str) -> str:
"""Force the psycopg2 driver for Postgres URLs.
Supabase/Heroku-style ``postgres://`` URLs are rejected by SQLAlchemy 2.x,
so both ``postgres://`` and bare ``postgresql://`` are rewritten to
``postgresql+psycopg2://``. Other schemes (e.g. ``sqlite://``) pass through.
"""
for prefix in ("postgres://", "postgresql://"):
if url.startswith(prefix):
return "postgresql+psycopg2://" + url[len(prefix) :]
return url
def _normalize_redis_url(url: str) -> str:
"""Ensure TLS Redis URLs carry ``ssl_cert_reqs``.
Celery refuses ``rediss://`` broker URLs without an explicit
``ssl_cert_reqs`` query parameter. Upstash presents a valid public
certificate, so ``required`` (full verification) is the safe default.
"""
if not url.startswith("rediss://"):
return url
parts = urlsplit(url)
query = dict(parse_qsl(parts.query))
query.setdefault("ssl_cert_reqs", "required")
return urlunsplit(parts._replace(query=urlencode(query)))
class Settings(BaseSettings):
"""Typed runtime configuration."""
model_config = SettingsConfigDict(
env_file=REPO_ROOT / ".env",
env_file_encoding="utf-8",
extra="ignore",
case_sensitive=True,
)
# --- Database / queue ---------------------------------------------------
DATABASE_URL: str = Field(..., description="SQLAlchemy URL (Supabase or local Postgres).")
REDIS_URL: str = Field(..., description="Redis URL (Upstash rediss:// or local redis://).")
CELERY_BROKER_URL: str = ""
CELERY_RESULT_BACKEND: str = ""
# --- App ------------------------------------------------------------------
ENVIRONMENT: Literal["development", "test", "production"] = "development"
SECRET_KEY: str = ""
CORS_ORIGINS: str = "http://localhost:3000"
REVIEW_RATE_LIMIT: int = Field(default=10, ge=1, le=10000)
REVIEW_RATE_WINDOW_SECONDS: int = Field(default=60, ge=1, le=86400)
RATE_LIMIT_MAX_CLIENTS: int = Field(default=10000, ge=1, le=100000)
MAX_REQUEST_BODY_BYTES: int = Field(default=65536, ge=1024, le=1048576)
TRUSTED_PROXY_CIDRS: str = ""
@field_validator("TRUSTED_PROXY_CIDRS")
@classmethod
def _trusted_proxies(cls, value: str) -> str:
for network in value.split(","):
if network.strip():
ip_network(network.strip(), strict=False)
return value.strip()
# --- Models -----------------------------------------------------------------
MODEL_DIR: Path = Path("data/models")
HF_MODEL_REPO: str = ""
HF_TOKEN: str = ""
@field_validator("DATABASE_URL")
@classmethod
def _db_url(cls, value: str) -> str:
return _normalize_database_url(value.strip())
@field_validator("REDIS_URL", "CELERY_BROKER_URL", "CELERY_RESULT_BACKEND")
@classmethod
def _redis_url(cls, value: str) -> str:
return _normalize_redis_url(value.strip())
@field_validator("MODEL_DIR")
@classmethod
def _model_dir(cls, value: Path) -> Path:
# Relative paths are anchored at the repo root so the API (run from
# `backend/`) and notebooks (run from the root) resolve the same folder.
return value if value.is_absolute() else (REPO_ROOT / value).resolve()
# --- Derived --------------------------------------------------------------
@property
def broker_url(self) -> str:
"""Celery broker; falls back to ``REDIS_URL``."""
return self.CELERY_BROKER_URL or self.REDIS_URL
@property
def result_backend(self) -> str:
"""Celery result backend; falls back to ``REDIS_URL``."""
return self.CELERY_RESULT_BACKEND or self.REDIS_URL
@property
def cors_origins(self) -> list[str]:
"""``CORS_ORIGINS`` split on commas, blanks removed."""
return [origin.strip() for origin in self.CORS_ORIGINS.split(",") if origin.strip()]
@property
def is_production(self) -> bool:
return self.ENVIRONMENT == "production"
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""Return the cached settings instance."""
return Settings() # values come from env
settings = get_settings()
|