refactor(config): 移除不必要的配置重新加载函数并优化设置更新逻辑

This commit is contained in:
snaily
2025-04-10 09:34:29 +08:00
parent af310ffb6b
commit 0f28173b0e
3 changed files with 31 additions and 71 deletions

View File

@@ -7,7 +7,6 @@ from typing import List, Any, Dict, Type
from pydantic import ValidationError
from pydantic_settings import BaseSettings
from dotenv import find_dotenv, load_dotenv
from sqlalchemy import insert, update, select
from app.core.constants import API_VERSION, DEFAULT_CREATE_IMAGE_MODEL, DEFAULT_FILTER_MODELS, DEFAULT_MODEL, DEFAULT_STREAM_CHUNK_SIZE, DEFAULT_STREAM_LONG_TEXT_THRESHOLD, DEFAULT_STREAM_MAX_DELAY, DEFAULT_STREAM_MIN_DELAY, DEFAULT_STREAM_SHORT_TEXT_THRESHOLD, DEFAULT_TIMEOUT
@@ -72,19 +71,6 @@ class Settings(BaseSettings):
# 创建全局配置实例
settings = Settings()
# 添加重新加载配置的函数
def reload_settings():
"""重新加载环境变量并更新配置"""
global settings
# 显式加载 .env 文件,覆盖现有环境变量
# find_dotenv() 会查找 .env 文件
load_dotenv(find_dotenv(), override=True)
settings = Settings()
# 可以在这里添加日志记录,确认配置已重新加载
# print("Settings reloaded") # 用于调试
# --- Initial Settings Synchronization ---
def _parse_db_value(key: str, db_value: str, target_type: Type) -> Any:
"""尝试将数据库字符串值解析为目标 Python 类型"""
try:

View File

@@ -3,15 +3,15 @@
"""
import datetime
import json
import os
from typing import Any, Dict, List
from dotenv import load_dotenv
from dotenv import find_dotenv, load_dotenv
from sqlalchemy import insert, update
from app.config.config import settings, reload_settings
from app.config.config import settings
from app.database.connection import database
from app.database.models import Settings
from app.config.config import Settings as ConfigSettings
from app.database.services import get_all_settings
from app.service.key.key_manager import get_key_manager_instance, reset_key_manager_instance
from app.log.logger import get_config_routes_logger
@@ -114,64 +114,35 @@ class ConfigService:
@staticmethod
async def reset_config() -> Dict[str, Any]:
"""
重置配置到默认值
重置配置:优先从系统环境变量加载,然后从 .env 文件加载,
更新内存中的 settings 对象,并刷新 KeyManager。
Returns:
Dict[str, Any]: 重置后的配置字典
"""
# 重新加载.env文件
load_dotenv(override=True)
# 重新加载配置对象以反映最新的环境变量
reload_settings()
logger.info("Settings object reloaded from environment variables.")
# 同步数据库中的配置到settings对象
await ConfigService._sync_db_config()
return await ConfigService.get_config()
@staticmethod
async def _sync_db_config() -> None:
"""
将.env文件中的配置项同步到数据库
"""
# 1. 重新加载配置对象,它应该处理环境变量和 .env 的优先级
_reload_settings()
logger.info("Settings object reloaded, prioritizing system environment variables then .env file.")
# 2. 重置并重新初始化 KeyManager
try:
# 获取.env文件中的所有配置项
env_values = dotenv_values(".env")
await ConfigService.update_config(env_values)
logger.info("Synced configuration to database")
await reset_key_manager_instance()
# 确保使用更新后的 settings 中的 API_KEYS
await get_key_manager_instance(settings.API_KEYS)
logger.info("KeyManager instance re-initialized with reloaded settings.")
except Exception as e:
logger.error(f"Failed to sync configuration to database: {str(e)}")
logger.error(f"Failed to re-initialize KeyManager during reset: {str(e)}")
# 根据需要决定是否抛出异常或继续
# 这里选择记录错误并继续
# 3. 返回更新后的配置
return await ConfigService.get_config()
# 添加dotenv_values函数
def dotenv_values(dotenv_path: str) -> Dict[str, str]:
"""
.env文件中读取配置项
Args:
dotenv_path: .env文件路径
Returns:
Dict[str, str]: 配置项字典
"""
if not os.path.exists(dotenv_path):
return {}
result = {}
with open(dotenv_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip()
# 去除引号
if value and value[0] == value[-1] and value[0] in ["'", '"']:
value = value[1:-1]
result[key] = value
return result
# 重新加载配置的函数
def _reload_settings():
"""重新加载环境变量并更新配置"""
# 显式加载 .env 文件,覆盖现有环境变量
load_dotenv(find_dotenv(), override=True)
# 更新现有 settings 对象的属性,而不是新建实例
for key, value in ConfigSettings().model_dump().items():
setattr(settings, key, value)

View File

@@ -2,6 +2,7 @@ import asyncio
from itertools import cycle
from typing import Dict
from app.config.config import settings
from app.log.logger import get_key_manager_logger
@@ -109,6 +110,8 @@ async def get_key_manager_instance(api_keys: list = None) -> KeyManager:
_singleton_instance = KeyManager(api_keys)
logger.info("KeyManager instance created.")
return _singleton_instance
async def reset_key_manager_instance():
"""重置 KeyManager 单例实例"""
global _singleton_instance