哪个网站做贷款推广,wordpress评论链接,音乐网站制作源代码,wordpress导航栏的文件在哪里前言
在开发和运维过程中#xff0c;配置管理是一个非常重要但经常被忽视的环节。常用的配置文件格式包括env、ini和yaml等#xff0c;它们非常适合模块级别的系统配置#xff0c;尤其是一些敏感信息的配置#xff0c;例如数据库连接字符串和密码等。但是#xff0c;对于…
前言
在开发和运维过程中配置管理是一个非常重要但经常被忽视的环节。常用的配置文件格式包括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_keyTrue, autoincrementTrue, commentID)cfg_name Column(String(128), nullableFalse, uniqueTrue, comment配置名称)cfg_value Column(TEXT, nullableTrue, comment配置值)cfg_desc Column(String(128), nullableTrue, comment配置描述)updated Column(TIMESTAMP, indexTrue, server_defaultfunc.now(), onupdatefunc.now(), nullableFalse, 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_dbTrue):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(fConfiguration 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(fConfiguration 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(fError 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(fConfigManager 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(fMax Connections: {max_connections})enable_feature config_mgr.bool(enable_feature, False)print(fEnable Feature: {enable_feature})custom_dict config_mgr.dict(custom_dict, {})print(fCustom Dict: {custom_dict})custom_list config_mgr.list(custom_list, [])print(fCustom List: {custom_list})except (KeyError, ValueError) as e:print(e)# 运行异步主函数
import asyncio
asyncio.run(main())
结语
通过上述代码示例我们展示了如何创建一个简单而有效的配置管理模块它能够动态加载和更新配置支持多种数据类型的转换并且在设计上注重高效和安全性。这个模块对于需要频繁更改业务逻辑配置而不希望重启服务的应用场景特别有用。
欢迎关注【程序员的开发手册】我们将继续分享更多实用的开发技巧和工具让您的开发之路更加顺畅。
