config.py 1.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. """
  2. scheduled-tasks 配置模块
  3. 当前复用 backend-client 的数据库配置:
  4. - 通过在 sys.path 中加入 ../backend-client 以导入 app.config.settings
  5. """
  6. import os
  7. import sys
  8. from pathlib import Path
  9. from typing import Optional
  10. def _ensure_backend_client_on_path() -> None:
  11. """
  12. 将 backend-client 根目录加入 sys.path,方便导入其 app 包。
  13. """
  14. here = Path(__file__).resolve()
  15. project_root = here.parents[2] # .../steam/
  16. backend_client = project_root / "backend-client"
  17. backend_client_str = str(backend_client)
  18. if backend_client_str not in sys.path:
  19. sys.path.insert(0, backend_client_str)
  20. _ensure_backend_client_on_path()
  21. try:
  22. # 复用 backend-client 的 settings(DATABASE_URL 等)
  23. from app.config import settings as _backend_settings # type: ignore
  24. except Exception: # pragma: no cover - 如果导入失败则回退到环境变量
  25. _backend_settings = None
  26. def get_database_url() -> str:
  27. """
  28. 获取数据库 URL:
  29. 1. 优先使用 backend-client 的 settings.DATABASE_URL
  30. 2. 否则回退到环境变量 CLIENT_DATABASE_URL
  31. """
  32. if _backend_settings and getattr(_backend_settings, "DATABASE_URL", None):
  33. return _backend_settings.DATABASE_URL # type: ignore[attr-defined]
  34. env_url: Optional[str] = os.getenv("CLIENT_DATABASE_URL")
  35. if not env_url:
  36. raise RuntimeError(
  37. "未找到数据库配置,请设置 backend-client 的 settings 或环境变量 CLIENT_DATABASE_URL"
  38. )
  39. return env_url