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

前言

在开发和运维过程中，配置管理是一个非常重要但经常被忽视的环节。常用的配置文件格式包括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())