86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import yaml
|
||
|
|
from pydantic import BaseModel, computed_field, Field
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
|
||
|
|
# 加载环境变量
|
||
|
|
load_dotenv()
|
||
|
|
|
||
|
|
|
||
|
|
class DatabaseConfig(BaseModel):
|
||
|
|
driver: str = "mysql+pymysql"
|
||
|
|
host: str
|
||
|
|
port: int = 3306
|
||
|
|
name: str
|
||
|
|
username: str
|
||
|
|
password: str
|
||
|
|
params: str = "characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowMultiQueries=true&rewriteBatchedStatements=true"
|
||
|
|
|
||
|
|
@computed_field
|
||
|
|
@property
|
||
|
|
def url(self) -> str:
|
||
|
|
"""生成数据库连接 URL"""
|
||
|
|
return f"{self.driver}://{self.username}:{self.password}@{self.host}:{self.port}/{self.name}?{self.params}"
|
||
|
|
|
||
|
|
|
||
|
|
class ServerConfig(BaseModel):
|
||
|
|
port: int
|
||
|
|
context_path: str = "/"
|
||
|
|
|
||
|
|
|
||
|
|
class AppConfig(BaseModel):
|
||
|
|
name: str = "wms_main"
|
||
|
|
max_file_size: int = 100 * 1024 * 1024 # 100MB
|
||
|
|
max_request_size: int = 1000 * 1024 * 1024 # 1000MB
|
||
|
|
|
||
|
|
|
||
|
|
class Settings(BaseModel):
|
||
|
|
"""全局配置模型"""
|
||
|
|
app: AppConfig
|
||
|
|
server: ServerConfig
|
||
|
|
database: DatabaseConfig
|
||
|
|
env: str = Field(default="development") # 当前环境名称
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def from_env(cls):
|
||
|
|
"""从环境变量加载配置"""
|
||
|
|
env_name = os.getenv("ENV", "development")
|
||
|
|
|
||
|
|
# 获取项目根目录
|
||
|
|
base_dir = Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
# 加载基础配置
|
||
|
|
base_path = base_dir / "config" / "base.yaml"
|
||
|
|
base_data = {}
|
||
|
|
if base_path.exists():
|
||
|
|
with open(base_path, "r") as f:
|
||
|
|
base_data = yaml.safe_load(f) or {}
|
||
|
|
|
||
|
|
# 加载环境特定配置
|
||
|
|
env_path = base_dir / "config" / f"{env_name}.yaml"
|
||
|
|
env_data = {}
|
||
|
|
if env_path.exists():
|
||
|
|
with open(env_path, "r") as f:
|
||
|
|
env_data = yaml.safe_load(f) or {}
|
||
|
|
|
||
|
|
# 深度合并配置
|
||
|
|
def deep_merge(base: dict, update: dict) -> dict:
|
||
|
|
"""递归合并两个字典"""
|
||
|
|
for key, value in update.items():
|
||
|
|
if isinstance(value, dict) and key in base and isinstance(base[key], dict):
|
||
|
|
base[key] = deep_merge(base[key], value)
|
||
|
|
else:
|
||
|
|
base[key] = value
|
||
|
|
return base
|
||
|
|
|
||
|
|
# 合并配置
|
||
|
|
merged_config = deep_merge(base_data, env_data)
|
||
|
|
merged_config["env"] = env_name # 添加环境名称
|
||
|
|
|
||
|
|
return cls(**merged_config)
|
||
|
|
|
||
|
|
|
||
|
|
# 创建全局配置对象
|
||
|
|
settings = Settings.from_env()
|