| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- """
- 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
|