Python实现一个强大的配置库

水深无声 2022-05-21 10:08 168阅读 0赞

应用程序经常要读取配置文件，还要验证输入是否有效、没有配置时使用默认值，于是我就想怎么简化这一流程

# 文件格式 #

首先为了方便用户直接改写，文件格式一定要是可读的。Python 标准库里可以读写可读配置的库有 configparser（ini文件）、json、xml，其中 ini、xml 文件不分类型，读取到的都是字符串，所以选择了 json。不过 JSON 有个缺点就是不支持注释，可以忽略这个缺点，提供好文档就行了

# 数据验证、默认值 #

我找到了一个 Python 库：[trafaret][]，它可以验证任何形式的数据并转换成需要的格式

这是把一个 `dict` 转换成 `datetime` 的例子：

import datetime
    import trafaret as t
    
    date = t.Dict({
        'year': t.Int,
        'month': t.Int,
        'day': t.Int
    }) >> (lambda d: datetime.datetime(**d))
    assert date.check({
       'year': 2012, 'month': 1, 'day': 12}) == datetime.datetime(2012, 1, 12)

首先声明个 `Dict` 结构，指定 key 和值的类型， `>>` 操作符可以指定自定义的转换函数，转换函数中可以抛出 `DataError` 错误， `date.check()` 可以返回验证并转换后的对象

`Dict` 的 key 除了字符串，还可以指定 `Key` 类型，`Key` 可以指定默认值以及把原名字转换成另一个名字：

>>> c = t.Dict({t.Key('un', 'default_user_name', True) >> 'user_name': t.String})
    >>> c.check({
       'un': 'Adam'})
    {
       'user_name': 'Adam'}
    >>> c.check({})
    {
       'user_name': 'default_user_name'}

这是声明一个可选的 key 叫 `'un'`，默认值是 `'default_user_name'`，`>>` 操作符指定读取后名字转换成 `'user_name'`，也可以在参数里指定转换的名字

# 实现 Config 类 #

trafaret 转换后得到的是 `dict`，`dict` 用的是字符串类型的 key，这对编辑器不友好，写代码时没有自动补全，而且想改名时也没办法自动批量修改，所以我打算用访问属性的方式访问配置（用 `.` 操作符访问）

受到各种 ORM 框架的启发，我打算在类属性里声明配置的字段，写法如下：

class NestedConfig(config):
        pass
    
    
    class MyConfig(Config):
        bool_field = {OptionalKey(True): t.Bool}
        int_field = {OptionalKey(123): t.Int}
        nested_config = {OptionalKey({}): NestedConfig}

## OptionalKey ##

这其实就是指定了一些默认参数的 Key 类，这样我们在参数里就只用写默认值了，转换后的名字由类属性的名字指定

class OptionalKey(t.Key):
        def __init__(self, default, name=None):
            super().__init__(name, default, True)

## 将声明的字段转换成 Dict ##

用 Python 的元类可以很容易实现这一点，我们在创建类时扫描声明的字段，转换成 `Dict` 并保存在 `__struct__` 类属性里。这里还实现了 `Config` 的继承，只要在创建 `__struct__` 的时候把基类已经创建好的 `__struct__` 合并进去就行了，注意基类不能覆盖子类的字段

class ConfigMeta(type):
        """用来创建Config类的__struct__ """
    
        def __new__(mcs, name, bases: Tuple[type, ...], namespace: Dict[str, Any]):
            # 添加此类的配置
            fields = set()
            cls_struct = {}
            for key, value in list(namespace.items()):
                if type(value) is dict and len(value) == 1:
                    field_key, field_checker = list(value.items())[0]
                    if isinstance(field_key, t.Key):
                        del namespace[key]
    
                        field_key: t.Key
                        # 默认name和类属性名一样
                        if field_key.name is None:
                            field_key.name = key
                        # 统一设置to_name，以后就不用判断to_name是否为None了
                        if field_key.to_name is None:
                            field_key.to_name = key
                        assert field_key.to_name == key, f'{field_key.to_name} != {key} 配置key.to_name必须和类属性名一样'
    
                        fields.add(field_key.to_name)
                        cls_struct[field_key] = field_checker
    
            # 添加基类的配置
            base_keys: List[t.Key] = []
            for base in bases:
                if issubclass(base, Config):
                    for key in base.__struct__.keys:
                        # 基类有，此类没有的配置
                        if key.to_name not in fields:
                            base_keys.append(key)
                            fields.update(key.to_name)
    
            cls = type.__new__(mcs, name, bases, namespace)
            cls.__struct__ = t.Dict(cls_struct, *base_keys)
            return cls

## Config 类的构造函数 ##

构造函数接受一个 `dict`，把它用 `__struct__` 转换后赋值到实例属性。这里顺便支持了传入关键词参数构造 `Config`。还要注意的是在 `check()` 之前要把未知的 key 移除，否则会出错

因为参数是 `dict`，这里支持了嵌套 `Config`。原理是 trafaret 的 checker 接受原始数据类型并返回转换后的数据类型，而 Python 里的类可以当做一个工厂函数，所以把 `Config` 类当做 checker 就可以实现嵌套了

class Config(metaclass=ConfigMeta):
    
        # 配置结构，见trafaret文档
        __struct__: t.Dict
    
        def __init__(self, raw: dict=None, **kwargs):
            # 支持传入dict或用关键字参数
            if raw is None:
                raw = {}
            raw = dict(raw, **kwargs)
    
            # 移除未知的key，防止check()出错
            known_keys = {key.name for key in self.__struct__.keys}
            for key in list(raw.keys()):
                if key not in known_keys:
                    del raw[key]
    
            raw = self.__struct__.check(raw)
            for key, value in raw.items():
                setattr(self, key, value)

## Config 转换回 dict ##

保存时还要把 `Config` 转换回 `dict`，JSON 才支持。由于支持了嵌套 `Config`，这里还要把底层的 `Config` 也转换成 `dict`

这里对于其他 JSON 不支持的数据类型没有做处理，所以 `Config` 里只能含有 JSON 支持的类型，否则不能保存。如果要使用其他数据类型可以用 `@property`

def to_dict(self):
            res = {key.to_name: getattr(self, key.to_name) for key in self.__struct__.keys}
            # 遍历dict和list组成的树，把Config转为dict
            queue: List[Union[dict, list]] = [res]
            while queue:
                node = queue.pop(0)
                for index, value in (node.items() if isinstance(node, dict)
                                     else enumerate(node)):
                    if isinstance(value, Config):
                        node[index] = value.to_dict()
                    elif isinstance(value, (dict, list)):
                        queue.append(value)
            return res

## 完整源码 ##

之后加上读取和保存配置就完成了，下面是完整源码：

# -*- coding: utf-8 -*-
    
    """配置模块 """
    
    import json
    from typing import Union, List, Dict, Any, Tuple
    
    import trafaret as t
    
    
    class OptionalKey(t.Key):
        def __init__(self, default, name=None):
            super().__init__(name, default, True)
    
    
    class ConfigMeta(type):
        """用来创建Config类的__struct__ """
    
        def __new__(mcs, name, bases: Tuple[type, ...], namespace: Dict[str, Any]):
            # 添加此类的配置
            fields = set()
            cls_struct = {}
            for key, value in list(namespace.items()):
                if type(value) is dict and len(value) == 1:
                    field_key, field_checker = list(value.items())[0]
                    if isinstance(field_key, t.Key):
                        del namespace[key]
    
                        field_key: t.Key
                        # 默认name和类属性名一样
                        if field_key.name is None:
                            field_key.name = key
                        # 统一设置to_name，以后就不用判断to_name是否为None了
                        if field_key.to_name is None:
                            field_key.to_name = key
                        assert field_key.to_name == key, f'{field_key.to_name} != {key} 配置key.to_name必须和类属性名一样'
    
                        fields.add(field_key.to_name)
                        cls_struct[field_key] = field_checker
    
            # 添加基类的配置
            base_keys: List[t.Key] = []
            for base in bases:
                if issubclass(base, Config):
                    for key in base.__struct__.keys:
                        # 基类有，此类没有的配置
                        if key.to_name not in fields:
                            base_keys.append(key)
                            fields.update(key.to_name)
    
            cls = type.__new__(mcs, name, bases, namespace)
            cls.__struct__ = t.Dict(cls_struct, *base_keys)
            return cls
    
    
    class Config(metaclass=ConfigMeta):
        """表示一个配置文件，可以以属性方式访问配置，支持继承 配置类型最好都是JSON支持的类型，如要使用其他类型可以用property 配置以类属性的方式声明，如： name = {OptionalKey(0): t.Int} 支持嵌套，如： nested_config = {OptionalKey({}): Config} 或者 nested_config = {OptionalKey({}): t.Dict >> Config} """
    
        # 配置结构，见trafaret文档
        __struct__: t.Dict
    
        def __init__(self, raw: dict=None, **kwargs):
            # 支持传入dict或用关键字参数
            if raw is None:
                raw = {}
            raw = dict(raw, **kwargs)
    
            # 移除未知的key，防止check()出错
            known_keys = {key.name for key in self.__struct__.keys}
            for key in list(raw.keys()):
                if key not in known_keys:
                    del raw[key]
    
            raw = self.__struct__.check(raw)
            for key, value in raw.items():
                setattr(self, key, value)
    
        def __repr__(self):
            return repr(self.to_dict())
    
        @classmethod
        def from_file(cls, filename):
            """如果文件不存在则使用默认配置 """
            try:
                with open(filename, encoding='utf-8') as f:
                    return cls(json.load(f))
            except FileNotFoundError:
                return cls()
    
        @classmethod
        def from_str(cls, s):
            return cls(json.loads(s))
    
        def to_dict(self):
            res = {key.to_name: getattr(self, key.to_name) for key in self.__struct__.keys}
            # 遍历dict和list组成的树，把Config转为dict
            queue: List[Union[dict, list]] = [res]
            while queue:
                node = queue.pop(0)
                for index, value in (node.items() if isinstance(node, dict)
                                     else enumerate(node)):
                    if isinstance(value, Config):
                        node[index] = value.to_dict()
                    elif isinstance(value, (dict, list)):
                        queue.append(value)
            return res
    
        def save(self, filename):
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)

# 使用方法 #

class BaseConfig(Config):
        # 这个字段会被子类覆盖
        override_field = {OptionalKey(1): t.Int}
        # 这个字段会被继承
        inherited_field = {OptionalKey(2): t.Int}
    
    
    class NestedConfig(Config):
        test = {OptionalKey(3): t.Int}
    
    
    class MyConfig(BaseConfig):
        # 覆盖基类的字段
        override_field = {OptionalKey(True): t.Bool}
        # 嵌套Config
        nested_config: NestedConfig = {OptionalKey({}): NestedConfig}
        nested_config_list: List[NestedConfig] = {OptionalKey([{
       'test': 123}]):
                                                  t.List(NestedConfig)}
        # 验证值是否在[1, 5]内
        int_field = {OptionalKey(1): t.Int(1, 5)}
        # 验证值是否在{1, 2, 3}内
        int_field2 = {OptionalKey(1): t.Int >>
                      (lambda x: x if x in (1, 2, 3) else t.DataError('int_field2不在{1, 2, 3}内'))}
    
    
    # 构造Config
    config = MyConfig({
       'int_field': 3, 'nested_config': {
       'test': 0}})
    # {'inherited_field': 2, 'override_field': True,
    # 'nested_config': {'test': 0}, 'nested_config_list': [{'test': 123}],
    # 'int_field': 3, 'int_field2': 1}
    print(repr(config))
    # 用关键字参数构造
    config = MyConfig(int_field=4)
    # {'inherited_field': 2, 'override_field': True,
    # 'nested_config': {'test': 3}, 'nested_config_list': [{'test': 123}],
    # 'int_field': 4, 'int_field2': 1}
    print(repr(config))
    # 以属性方式访问配置
    print(config.int_field)  # 4
    print(config.nested_config.test)  # 3
    # 检查类型和范围
    # trafaret.dataerror.DataError: {'override_field': DataError(value should be True or False)}
    # config = MyConfig({'override_field': 1})
    # trafaret.dataerror.DataError: {'int_field2': DataError(int_field2不在{1, 2, 3}内)}
    # config = MyConfig({'int_field2': 4})

[trafaret]: https://trafaret.readthedocs.io/en/latest/intro.html