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

前言

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

系统配置表

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

from sqlalchemy import (
    TEXT,
    TIMESTAMP,
    Column,
    Integer,
    String,
    func,
)

from app.models.base import Base, BaseMixin

class 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, Callable

import orjson
from app.models.sys_config import SysConfig

T = 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 = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(ConfigManager, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        self.configs: Dict[str, str] = {}
        self.initialized = False

    async 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 = True
        print("Configurations loaded into memory.")

    async def update_config(self, key: str, value: str, description: str = '', write_to_db=True):
        self.configs[key] = value
        if write_to_db:
            record = {'cfg_name': key, 'cfg_value': value}
            if description:
                record['cfg_desc'] = description
            await 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 method
        raise AttributeError(f"'ConfigManager' object has no attribute '{item}'")

使用示例

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

from app.services import config_services

async 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())

结语

通过上述代码示例，我们展示了如何创建一个简单而有效的配置管理模块，它能够动态加载和更新配置，支持多种数据类型的转换，并且在设计上注重高效和安全性。这个模块对于需要频繁更改业务逻辑配置而不希望重启服务的应用场景特别有用。

欢迎关注【程序员的开发手册】，我们将继续分享更多实用的开发技巧和工具，让您的开发之路更加顺畅。