from pathlib import Path from typing import Literal from sqlalchemy.engine import URL from pydantic import BaseModel from pydantic_yaml import parse_yaml_raw_as 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 Settings(BaseModel): """应用配置""" database: DatabaseConfig @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, ) def load_settings(config_path: str = "config/settings.yaml") -> Settings: """加载 YAML 配置文件""" # Resolve relative to the project root (where config/ directory exists) path = Path(config_path) if not path.is_absolute(): # __file__ is config/settings.py, so parent.parent gives us project root project_root = Path(__file__).resolve().parent.parent path = project_root / config_path if not path.exists(): raise FileNotFoundError( f"配置文件不存在: {config_path}\n" f"请确保文件存在于项目根目录或指定正确路径" ) # Use UTF-8 encoding to avoid Windows GBK encoding issues with open(path, "r", encoding="utf-8") as f: return parse_yaml_raw_as(Settings, f) # 全局配置单例 settings = load_settings()