111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
from dataclasses import dataclass
|
||
from functools import lru_cache
|
||
import os
|
||
from typing import Any
|
||
|
||
try:
|
||
from pathlib import Path as _Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
# 显式指定 backend 目录下的 .env，并强制覆盖已有环境变量，
|
||
# 确保远程数据库配置生效，不被本地默认值或残留环境变量影响。
|
||
_env_path = _Path(__file__).resolve().parent.parent.parent / ".env"
|
||
load_dotenv(dotenv_path=_env_path, override=True)
|
||
except ImportError:
|
||
pass
|
||
|
||
|
||
def _int_env(name: str, default: int) -> int:
|
||
raw = os.getenv(name)
|
||
if raw is None or raw == "":
|
||
return default
|
||
return int(raw)
|
||
|
||
|
||
def _list_env(name: str, default: list[str]) -> list[str]:
|
||
raw = os.getenv(name)
|
||
if raw is None or raw.strip() == "":
|
||
return default
|
||
return [item.strip() for item in raw.split(",") if item.strip()]
|
||
|
||
|
||
def _bool_env(name: str, default: bool) -> bool:
|
||
raw = os.getenv(name)
|
||
if raw is None or raw.strip() == "":
|
||
return default
|
||
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
||
|
||
|
||
def docs_kwargs(enabled: bool) -> dict[str, Any]:
|
||
"""Swagger UI / ReDoc / OpenAPI schema 路由开关。
|
||
|
||
关闭时 FastAPI 不注册 /docs、/redoc、/openapi.json，访问一律返回 404，
|
||
避免未授权访问泄露 API 结构。
|
||
"""
|
||
if enabled:
|
||
return {}
|
||
return {"docs_url": None, "redoc_url": None, "openapi_url": None}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Settings:
|
||
app_name: str = os.getenv("APP_NAME", "YG Zhilian API")
|
||
app_env: str = os.getenv("APP_ENV", "local")
|
||
route_prefix: str = os.getenv("MODELTF_ROUTE_PREFIX", "/modelTF")
|
||
app_mode: str = os.getenv("APP_MODE", "local")
|
||
database_url: str = os.getenv("DATABASE_URL", "postgresql+psycopg://yg_ft:change_me@localhost:15432/yg_ft")
|
||
cors_allow_origins: list[str] = None # type: ignore[assignment]
|
||
compute_mode: str = os.getenv("COMPUTE_MODE", "real")
|
||
compute_status_sync_mode: str = os.getenv("COMPUTE_STATUS_SYNC_MODE", "polling")
|
||
compute_poll_interval_seconds: int = _int_env("COMPUTE_POLL_INTERVAL_SECONDS", 3)
|
||
compute_request_timeout_seconds: int = _int_env("COMPUTE_REQUEST_TIMEOUT_SECONDS", 5)
|
||
minio_enabled: bool = _bool_env("MINIO_ENABLED", False)
|
||
# MinIO is an independent service and may run on another host. The
|
||
# endpoint must therefore be reachable from the Backend container.
|
||
minio_endpoint: str = os.getenv("MINIO_ENDPOINT", "http://host.docker.internal:19000")
|
||
minio_access_key: str = os.getenv("MINIO_ACCESS_KEY", "minioadmin")
|
||
minio_secret_key: str = os.getenv("MINIO_SECRET_KEY", "minioadmin")
|
||
minio_bucket: str = os.getenv("MINIO_BUCKET", "yg-ft-resources")
|
||
minio_secure: bool = _bool_env("MINIO_SECURE", False)
|
||
storage_wait_seconds: int = _int_env("STORAGE_WAIT_SECONDS", 300)
|
||
storage_check_interval_seconds: int = _int_env("STORAGE_CHECK_INTERVAL_SECONDS", 10)
|
||
compute_service_token: str = os.getenv("COMPUTE_SERVICE_TOKEN", "")
|
||
log_level: str = os.getenv("LOG_LEVEL", "INFO")
|
||
log_dir: str = os.getenv("LOG_DIR", "./logs")
|
||
log_file_prefix: str = os.getenv("LOG_FILE_PREFIX", "backend")
|
||
log_error_file_prefix: str = os.getenv("LOG_ERROR_FILE_PREFIX", "error")
|
||
log_max_bytes: int = _int_env("LOG_MAX_BYTES", 20 * 1024 * 1024)
|
||
log_retention_days: int = _int_env("LOG_RETENTION_DAYS", 10)
|
||
enable_docs: bool = None # type: ignore[assignment]
|
||
|
||
def __post_init__(self) -> None:
|
||
object.__setattr__(
|
||
self,
|
||
"cors_allow_origins",
|
||
_list_env(
|
||
"CORS_ALLOW_ORIGINS",
|
||
[
|
||
"http://localhost:16801",
|
||
"http://127.0.0.1:16801",
|
||
"http://localhost:17861",
|
||
"http://127.0.0.1:17861",
|
||
],
|
||
),
|
||
)
|
||
# Swagger UI / ReDoc / OpenAPI 文档路由开关：
|
||
# 未显式配置 ENABLE_DOCS 时，仅本地/开发环境开放，生产环境默认关闭，
|
||
# 避免未授权访问泄露 API 结构。从运行时环境读取 APP_ENV，而非类定义时
|
||
# 缓存的默认值，保证生产默认关闭始终生效且便于测试。
|
||
object.__setattr__(
|
||
self,
|
||
"enable_docs",
|
||
_bool_env("ENABLE_DOCS", os.getenv("APP_ENV", "local") != "prod"),
|
||
)
|
||
|
||
|
||
@lru_cache
|
||
def get_settings() -> Settings:
|
||
return Settings()
|
||
|