"""数据库配置加载。 与 services/fastapi(CargoTrace 后端)的 config/settings.py 保持同构, 复用同一份 config/settings.yaml 凭据。 """ from pathlib import Path from typing import Literal from pydantic import BaseModel from pydantic_yaml import parse_yaml_raw_as from sqlalchemy.engine import URL class SqlServerConfig(BaseModel): """SQL Server 连接配置""" host: str port: int = 1433 database: str username: str password: str driver: str = "{ODBC Driver 18 for SQL Server}" trust_server_certificate: str = "yes" class PostgreSqlConfig(BaseModel): """PostgreSQL 连接配置""" host: str port: int = 5432 database: str username: str password: str class DatabaseConfig(BaseModel): """数据库配置""" active: Literal["sql_server", "postgresql"] = "sql_server" sql_server: SqlServerConfig postgresql: PostgreSqlConfig | None = None class ReportConfig(BaseModel): """报表配置""" # 调试开关:True 时给所有文本元素加边框,便于核对每个元素的实际占位。 debug_border: bool = False class Settings(BaseModel): """应用配置""" database: DatabaseConfig report: ReportConfig = ReportConfig() @property def database_url(self) -> URL: """构建数据库连接 URL""" if self.database.active == "postgresql": return self._postgresql_url() return self._sql_server_url() def _sql_server_url(self) -> URL: conf = self.database.sql_server return URL.create( "mssql+pyodbc", username=conf.username, password=conf.password, host=conf.host, port=conf.port, database=conf.database, query={ "driver": conf.driver.strip("{}"), "TrustServerCertificate": conf.trust_server_certificate, }, ) def _postgresql_url(self) -> URL: conf = self.database.postgresql if conf is None: raise ValueError("已选择 postgresql,但未配置 database.postgresql") return URL.create( "postgresql+psycopg", username=conf.username, password=conf.password, host=conf.host, port=conf.port, database=conf.database, ) # 项目根目录:core/settings.py 的 parent.parent PROJECT_ROOT = Path(__file__).resolve().parent.parent def load_settings(config_path: str = "config/settings.yaml") -> Settings: """加载 YAML 配置文件(相对路径基于项目根目录解析)。""" path = Path(config_path) if not path.is_absolute(): path = PROJECT_ROOT / config_path if not path.exists(): raise FileNotFoundError( f"配置文件不存在: {config_path}\n" f"请复制 config/settings.example.yaml 为 config/settings.yaml 并填入凭据。" ) with open(path, "r", encoding="utf-8") as f: return parse_yaml_raw_as(Settings, f) settings = load_settings()