""" scheduled-tasks 配置模块 当前复用 backend-client 的数据库配置: - 通过在 sys.path 中加入 ../backend-client 以导入 app.config.settings """ import os import sys from pathlib import Path from typing import Optional def _ensure_backend_client_on_path() -> None: """ 将 backend-client 根目录加入 sys.path,方便导入其 app 包。 """ here = Path(__file__).resolve() project_root = here.parents[2] # .../steam/ backend_client = project_root / "backend-client" backend_client_str = str(backend_client) if backend_client_str not in sys.path: sys.path.insert(0, backend_client_str) _ensure_backend_client_on_path() try: # 复用 backend-client 的 settings(DATABASE_URL 等) from app.config import settings as _backend_settings # type: ignore except Exception: # pragma: no cover - 如果导入失败则回退到环境变量 _backend_settings = None def get_database_url() -> str: """ 获取数据库 URL: 1. 优先使用 backend-client 的 settings.DATABASE_URL 2. 否则回退到环境变量 CLIENT_DATABASE_URL """ if _backend_settings and getattr(_backend_settings, "DATABASE_URL", None): return _backend_settings.DATABASE_URL # type: ignore[attr-defined] env_url: Optional[str] = os.getenv("CLIENT_DATABASE_URL") if not env_url: raise RuntimeError( "未找到数据库配置,请设置 backend-client 的 settings 或环境变量 CLIENT_DATABASE_URL" ) return env_url