当前位置：首页 > news >正文

新手教学系列——简单的服务配置项集中管理

news 2025/7/22 22:38:16

前言

在开发和运维过程中，配置管理是一个非常重要但经常被忽视的环节。常用的配置文件格式包括env、ini和yaml等，它们非常适合模块级别的系统配置，尤其是一些敏感信息的配置，例如数据库连接字符串和密码等。但是，对于系统业务级别的配置，通常要求不需要重启服务即可更新，这就是我们今天要介绍的简单配置管理模块的意义所在。

系统配置表

首先，我们需要一个数据库表来存储配置项。这个表包括配置名称、配置值和配置描述等信息。以下是一个使用SQLAlchemy定义的配置表模型：

from sqlalchemy import (TEXT,TIMESTAMP,Column,Integer,String,func,
)from app.models.base import Base, BaseMixinclass SysConfig(Base, BaseMixin):__tablename__ = 'sys_configs'__table_args__ = {"comment": "系统配置表"}id = Column(Integer, primary_key=True, autoincrement=True, comment='ID')cfg_name = Column(String(128), nullable=False, unique=True, comment='配置名称')cfg_value = Column(TEXT, nullable=True, comment='配置值')cfg_desc = Column(String(128), nullable=True, comment='配置描述')updated = Column(TIMESTAMP, index=True, server_default=func.now(), onupdate=func.now(), nullable=False, comment='更新时间')

配置管理类

接下来，我们需要一个配置管理类来加载和更新配置。这个类将会以单例模式运行，确保所有地方使用的配置都是一致的，并且在首次创建实例时自动加载所有配置项。我们使用异步操作来确保数据库操作的高效性。

import json
from typing import Any, Dict, Optional, Type, TypeVar, Callableimport orjson
from app.models.sys_config import SysConfigT = TypeVar('T')# 获取配置管理单例
async def get_config_manager():config_mgr = ConfigManager()if not config_mgr.initialized:await config_mgr.load_configs()return config_mgr# 配置管理类
class ConfigManager:_instance = Nonedef __new__(cls, *args, **kwargs):if cls._instance is None:cls._instance = super(ConfigManager, cls).__new__(cls)return cls._instancedef __init__(self):self.configs: Dict[str, str] = {}self.initialized = Falseasync def load_configs(self):cfg_rows = await SysConfig.get_all_async()for row in cfg_rows:self.configs[row['cfg_name']] = row['cfg_value']self.initialized = Trueprint("Configurations loaded into memory.")async def update_config(self, key: str, value: str, description: str = '', write_to_db=True):self.configs[key] = valueif write_to_db:record = {'cfg_name': key, 'cfg_value': value}if description:record['cfg_desc'] = descriptionawait SysConfig.upsert_async(records=[record], update_keys=['cfg_name'])print(f"Configuration updated: {key} = {value}")def _convert(self, key: str, type_: Type[T], default_value: Optional[T] = None) -> T:value = self.configs.get(key, default_value)if value is None:raise KeyError(f"Configuration key '{key}' not found and no default value provided.")try:if type_ == bool:return type_(value.lower() in ['true', '1', 'yes'])elif type_ == dict or type_ == list:return orjson.loads(value)return type_(value)except (ValueError, TypeError, json.JSONDecodeError) as e:raise ValueError(f"Error converting configuration value '{value}' to type {type_.__name__}: {e}")def __getattr__(self, item: str) -> Callable[[str, Optional[Any]], Any]:supported_types = {'int': int,'float': float,'bool': bool,'str': str,'dict': dict,'list': list,'json': dict,}if item in supported_types:def method(key: str, default_value: Optional[Any] = None) -> Any:return self._convert(key, supported_types[item], default_value)return methodraise AttributeError(f"'ConfigManager' object has no attribute '{item}'")

使用示例

现在，我们已经有了一个完整的配置管理模块，让我们看一下如何在实际应用中使用它。以下是一个示例代码，展示了如何获取配置管理器并使用它来获取和更新配置项。

from app.services import config_servicesasync def main():# 获取配置管理器单例config_mgr = await config_services.get_config_manager()# 更新配置await config_mgr.update_config('max_connections', '100', '最大连接数')await config_mgr.update_config('enable_feature', 'true', '启用新功能')await config_mgr.update_config('custom_dict', '{"key": "value"}', '自定义字典')await config_mgr.update_config('custom_list', '["item1", "item2"]', '自定义列表')# 获取并转换配置值try:max_connections = config_mgr.int('max_connections', 10)print(f"Max Connections: {max_connections}")enable_feature = config_mgr.bool('enable_feature', False)print(f"Enable Feature: {enable_feature}")custom_dict = config_mgr.dict('custom_dict', {})print(f"Custom Dict: {custom_dict}")custom_list = config_mgr.list('custom_list', [])print(f"Custom List: {custom_list}")except (KeyError, ValueError) as e:print(e)# 运行异步主函数
import asyncio
asyncio.run(main())