112 lines
3.0 KiB
Python
112 lines
3.0 KiB
Python
"""数据库配置加载。
|
||
|
||
与 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()
|