中小型 Python 项目配置和数据层的最佳实践

AhFei 在编写自用程序的过程中,逐步形成了一套清晰好用的 Python配置文件管理 和数据管理的程序模式,分享以交流。适用于几千行以内的中小项目,目前用这个模式很是舒适。

程序穿过 config 和 data 开启核心逻辑运行

程序配置,也就是程序运行依赖的若干值,比如是否在生产环境,管理员用户的账号密码,限制资源占用的变量,数据库的配置等。

由于配置文件一般存储有机密信息,所以不能写在程序文件中,那样也不利于放在一处统一管理,可以用一个文件存储这些配置,并且用专门的配置类在程序运行时读取配置,供程序后续使用。

配置文件需要手动编写,因此首要的是结构清晰简洁、便于观察和编辑,往往只在启动时被读取一次,因此性能并不重要,所以推荐使用 YAML 格式编写。


至于数据管理,其实核心就是抽象出程序所需的所有数据读写操作或者进一步程序与外部的所有交互,在程序的真正业务代码中,通过调用数据层中的方法进行读写外部数据。

这样不仅将程序核心与外部隔离,解耦降低复杂度,便于并行开发;还能在换用外部依赖时,只需要修改数据层,程序其他部分完全不用改动;并且,能够为常用的操作专门编写方法,从而降低使用时的繁琐。

另有一点,在数据层中,对外部数据进行验证,确保进入程序核心的数据都是合法的,从而不必在核心代码中还要考虑这些异常。

程序结构

project
│
├── config_and_data_files
│   ├── config.yaml
│   └── some_data.json
│
├── .gitignore
├── preprocess.py
├── configHandle.py
├── dataHandle.py
└── main.py

说明:

  1. 配置类为 configHandle.py
  2. 数据类为 dataHandle.py
  3. 配置文件和数据文件放在 config_and_data_files 目录下,在 .gitignore 中添加忽略
  4. 至于 preprocess.py 也就是预处理,请往下看

preprocess.py

主要功能是

  1. 添加命令行参数解析,比如测试时指定测试用的配置文件位置,增加灵活性
  2. 实例化 configHandle 为 config
  3. 实例化 dataHandle 为 data
import argparse
from configHandle import Config

# 创建一个解析器
parser = argparse.ArgumentParser(description="Your script description")
# 添加你想要接收的命令行参数
parser.add_argument('--config', required=False, default='config_and_data_files/config.yaml', help='Config File Path')
# 解析命令行参数
args = parser.parse_args()

# 实例化 配置类
configfile = args.config
config = Config(configfile)

# 如果配置没出错,继续实例化数据类
from dataHandle import Data
data = Data(config)

一开始我是在所谓的入口文件中放上面代码,比如是 main.py,但这样容易产生循环引用,比如 main 引用的模块需要配置类中的全局参数。

并且从结构上看,配置和数据处理是任何一个模块都可能需要的,因此单独放入预处理文件中更合适。而且这部分代码很少改变,拿出来看不见更清爽。


如果你的程序本身只有几个文件,并且没有一个以上的数据文件,那么完全可以不创建 config_and_data_files 目录,直接把 config.yaml 放在项目根目录,然后修改 parser.add_argument 中的默认参数。或者,在启动程序时指定配置文件路径:

python main.py --config ./config.yaml

配置类

如果项目小,配置项很少,用 JSON 格式也可以。但是当结构复杂,参数较多时,JSON 的嵌套结构就会很混乱,不宜阅读。

YAML 的样式非常美观,编辑也很容易。TOML 相比就不怎么好看。我使用 ruamel.yaml 库,它能在修改后,还保持原来的格式(注释位置,空行,- 表示 或者 [] 表示这种)

python -m pip install ruamel.yaml

configHandle.py

思路是:

  1. 实例化时接收一个参数: 配置文件路径
  2. 编写一个函数读取配置文件,将其中参数保存为字典
  3. 实例化时调用方法,将字典中的参数都存储为变量,如 self.is_production = configs['is_production'],从而通过 config.is_production 获得这项参数
  4. 上面的这个方法如果在运行中再调用一次,就相当于重载了配置文件,因此我将这个方法直接命名为 reload()
import sys
import logging.config

from ruamel.yaml import YAML, YAMLError

class Config:
    def __init__(self, configs_path='./configs.yaml') -> None:
        self.yaml = YAML()
        self.configs_path = configs_path
        self.reload()

        # 用户可以不管,开发者可以改的配置放这里

    def _load_config(self) -> dict:
        """定义如何加载配置文件"""
        try:
            with open(self.configs_path, "r", encoding='utf-8') as fp:
                configs = self.yaml.load(fp)
            return configs
        except YAMLError as e:
            sys.exit(f"The config file is illegal as a YAML: {e}")
        except FileNotFoundError:
            sys.exit(f"The config does not exist")

    def reload(self) -> None:
        """将配置文件里的参数,赋予单独的变量,方便后面程序调用"""
        configs = self._load_config()
        # 对日志模块进行配置,不使用删除即可
        logging.config.dictConfig(configs["logging"])
        # 配置文件中的参数,手动提取出来
        self.is_production = configs['is_production']

使用案例:

  1. 在入口文件中,通过 from preprocess import config 引入 config,主函数可以直接传 config 对象 给调度器之类的,或者只传 config.is_production 这样的具体变量,取决于程序结构和后续需要的参数
  2. 一个比较独立的模块要使用全局参数,如果依靠传递,那传递链的维护就很繁琐且容易出错,直接通过引入 config: from preprocess import config ,然后,用哪个变量,就 config.variable_name 即可
  3. 添加参数,比如 name=vfly2 时,只要三步
    1. 在 config_and_data_files/config.yaml 里添加 name: vfly2
    2. 然后在 Config 的 reload() 下分配变量名 self.name=configs['name']
    3. 通过 config.name 调用
  4. 运行时调用 config.reload() 重载配置

配置文件例子

is_production: false

logging:
  version: 1
  disable_existing_loggers: False

  formatters:
    simple:
      format: "%(asctime)s %(message)s"
      datefmt: "%Y-%m-%d %H:%M:%S"
    error:
      format: "%(asctime)s %(name)s %(levelname)s %(filename)s::%(funcName)s[%(lineno)d]:%(message)s"

  handlers:
    console:
      class: logging.StreamHandler
      level: INFO
      formatter: simple
      stream: ext://sys.stdout

    info_file_handler:
      class: logging.handlers.RotatingFileHandler
      level: INFO
      formatter: simple
      filename: info.log
      maxBytes: 10485760  # 10MB
      backupCount: 20
      encoding: utf8

    error_file_handler:
      class: logging.handlers.RotatingFileHandler
      level: ERROR
      formatter: error
      filename: error.log
      maxBytes: 10485760  # 10MB
      backupCount: 20
      encoding: utf8

  root:
    level: INFO
    handlers: [console, info_file_handler, error_file_handler]

日志模块的使用

# other_module.py
import logging

logger = logging.getLogger("module_name")
logger.info("输出日志")

可以为每个类都添加各自的日志实例,这样输出的日志带有类名,更容易定位问题代码。

import logging

class Example:
    def __init__(self):
        self.logger = logging.getLogger(self.__class__.__name__)

    def method_name(self):
        self.logger.info("...")

多配置文件

AhFei 的项目中,程序参数有 3 种:

  1. 用户级的参数。就是文章开头提到的,测试时会修改的
  2. 程序级的参数。普通用户不应该改动的,测试时保持不变的;比如日志配置,一些可能会失效的依赖项。
  3. 开发级的参数。剩下的参数,都应该直接写入到 Config 类的 __init__ 中,比如其他代码中用到了路径,或者表的名称,应该将其作为参数放在这里,这样程序中可变的量都汇集在一起,便于寻找和修改。
  4. 顺便一提,程序中用到的枚举量,个人感觉放在 DataHandle.py 更合适,因为它属于内部数据,而非外部传入的。

如果参数很多,可以使用两个配置文件,一个放用户级的参数,另一个放程序级,对上面 Config 稍加修改:

  1. 实例化时传参是一个字符串列表,包含所有配置文件的路径
  2. 加载配置文件时,遍历列表,返回一个可迭代对象
  3. 保存参数时,根据关键字划分给 user_configs(含有 user_configuration: true) 或 program_configs(含有 program_configuration: true),然后再提取参数给不同变量名。
import sys
import logging.config
from typing import Generator, Any

from ruamel.yaml import YAML, YAMLError

class Config:
    def __init__(self, configs_path: list[str]=["./configs.yaml", "./pgm_configs.yaml"]) -> None:
        self.yaml = YAML()
        self.configs_path = configs_path
        self.reload()

        # 用户可以不管,开发者可以改的配置放这里

    def _load_config(self) -> Generator[dict, Any, Any]:
        """定义如何加载配置文件"""
        for f in self.configs_path:
            try:
                with open(f, "r", encoding='utf-8') as fp:
                    configs = self.yaml.load(fp)
                yield configs
            except YAMLError as e:
                sys.exit(f"The config file is illegal as a YAML: {e}")
            except FileNotFoundError:
                sys.exit(f"The config does not exist")

    def reload(self) -> None:
        """将配置文件里的参数,赋予单独的变量,方便后面程序调用"""
        for i, configs in enumerate(self._load_config()):
            if configs.get("user_configuration"):
                user_configs = configs
            elif configs.get("program_configuration"):
                program_configs = configs
            else:
                sys.exit(f"{self.configs_path[i]} unknow configuration, lacking key for identify")

        # 默认无须用户改动的
        logging.config.dictConfig(program_configs["logging"])

        # 用户配置
        self.is_production = user_configs['is_production']

preprocess.py 也要稍加修改 parser.add_argument()

import argparse

parser = argparse.ArgumentParser(description="Your script description")
parser.add_argument('--config', action='append', required=False, 
                    default=None, 
                    help='Config Files Path')
args = parser.parse_args()
# 如果用户没有提供 --config 参数,那么将其设置为默认值
if args.config is None:
    args.config = ['./config_and_data_files/config.yaml', './config_and_data_files/pgm_config.yaml']

configfile = args.config
print(configfile)

...

配置文件例子

config_and_data_files/config.yaml

user_configuration: true
is_production: true

config_and_data_files/pgm_config.yaml

program_configuration: true

logging:
  version: 1
  # 省略剩余的

如果不使用默认参数,那在启动时,使用下面方式传入多个配置文件

python main.py --config ./config.yaml --config ./pgm_config.yaml

数据层

由人编写的,用 YAML 更合适。纯运行中产生的数据,很少查看不会手动编辑的,使用 JSON

dataHandle.py

下面的配置是同步代码下的思考,异步代码下,比如多个用户并发运行,只能各自实例化 data,否则数据会混在一起。

外部数据有三种,这也是数据层要操作的对象

  1. 那些在程序运行时,不会修改内容的,直接读取即可
  2. 运行时,会由外部改变的,需要考虑重载
  3. 运行时,会由程序自身改动的,需要考虑便捷修改和保存数据到硬盘

简单例子仅作示范,因此通过一个类来囊括全部数据层的操作,思路:

  1. 实例化时接收 config (配置类的实例),方便根据配置加载数据
  2. 实例化时要确保必要的数据文件存在,如果不存在就创建,后面读写时就不必考虑了。
  3. 实例化时直接加载运行中不会变化的数据,比如从高德获取天气需要用城市的编码,这种对应关系在启动时直接读入内存即可
  4. 如果数据文件是用户编写的,还要检查格式,保证后续程序不需要考虑不合法的问题
  5. 为常用的操作编写便捷方法,比如保存所有数据到各自文件,重载某个数据。

Data 类很灵活,把握核心思想编写便捷方法应该是主要工作。业务代码中不要出现 with open 或者 r.push() 这种直接的操作,而是调用 data.reload(), data.push_to_list() 这种。

import os
import json
import logging
from enum import Enum

from ruamel.yaml import YAML

# 如果需要,可以定义枚举量,避免魔法数字
class TaskType(Enum):
    """定义任务的类型"""
    SIMPLE_RT = 0
    ROLL_POLLING = 1
    LONG_RT = 2

class Data:
    yaml = YAML()

    def __init__(self, config) -> None:
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)
        self._make_sure_file_exist()
        self._load_file()

        # 如果使用数据库,这里可以创建对象,然后创建一些常用操作的方法

    def _load_file(self) -> None:
        """加载运行中不会变化的数据和其他数据"""
        self.city_adcode = self.reload(self.config.city_adcode_file)
        # 会变化的
        self.version_data = self.reload(self.config.version_file_path)
        self.original_hash4version = hash(json.dumps(self.version_data, sort_keys=True))

    def save_all(self):
        """可以为结束程序时编写一个方法,保存所有数据到文件,应该很少使用"""
        # 因为用的不多,所以方法也很原始,手动编辑要保存的变量和对应的文件路径,然后一个个保存
        file_list = []
        data_list = []
        for file, data in zip(file_list, data_list):
            Data.save(data, file)

    def save_version_deque(self):
        """专为保存版本信息的方法,每次调度之后都会保存最新的"""
        # 计算哈希值,判断是否有变化
        new_version = hash(json.dumps(self.version_data, sort_keys=True))
        if self.original_hash != new_version:
            # 有更新
            self.logger.info("当前已下载的最新版本信息已经改变,保存到文件中")
            with open(self.config.version_file_path, 'w', encoding='utf-8') as f:
                json.dump(self.version_data, f, ensure_ascii=False)
            self.original_hash = new_version
        else:
            self.logger.info("当前已下载的最新版本信息未发生改变")

    def _make_sure_file_exist(self) -> None:
        """有些数据文件,要确保存在,填充符合规范的样例内容,后面程序才能无须再判断"""
        # 这里主要是列举需要保证的内容,确保存在,并初始化内容的函数是 _check_file_exist
        # 不存在就放入样例内容
        sample_version = {"sample_project": "v0.01"}
        self._check_file_exist(self.config.version_file_path, sample_version)

    def _check_file_exist(self, file_path, example_content) -> None:
        if not os.path.exists(file_path):
            self.save(example_content, file_path)
        else:   # 存在则判断是否符合格式,这个还需要编写格式的规则,比较复杂就省略了
            self._make_sure_file_exist(file_path)

    def _make_sure_file_format(self) -> None:
        """确保数据文件的格式正确,后面程序才能无须再判断"""
        pass

    @classmethod
    def reload(cls, file_path):
        """初始化时用来加载某个文件的内容,运行中方便重载"""
        suffix_name = os.path.splitext(file_path)[1]   # 获取后缀名
        with open(file_path, 'r', encoding='utf-8') as f:
            if suffix_name == '.yaml':
                content = cls.yaml.load(f)
            elif suffix_name == '.json':
                content = json.load(f)
            else:
                raise Exception("Unknow file format")

        return content

    @classmethod
    def save(cls, data_in_m, file_path):
        suffix_name = os.path.splitext(file_path)[1]   # 获取后缀名
        with open(file_path, 'w', encoding='utf-8') as f:
            if suffix_name == '.yaml':
                cls.yaml.dump(data_in_m, f)
            elif suffix_name == '.json':
                json.dump(data_in_m, f)
            else:
                raise Exception("Unknow file format")

如何使用:

  1. 在入口文件中,通过 from preprocess import data 引入 data,推荐主函数直接把 data 传给其他实例或函数
  2. 后面程序中要读写数据时,就通过 data.method() 操作即可,若要增加操作,就在 Data 类中编写,然后通过 data.method() 调用,完全不需要再传递新参数

读取环境变量

Gunicorn、Uvicorn 启动情况下,就不能在 preprocess 中引入 argparse 了,因为像是 Uvicorn 本身也读取命令行参数,不方便让它再传递一些参数给后面的程序,这时候可以用环境变量代替。

preprocess.py

单配置文件

import os
from configHandle import Config

configfile = os.getenv("PROJECT_CONFIG_FILE", default='config_and_data_files/config.yaml')
absolute_configfile = os.path.join(os.getcwd(), configfile)

config = Config(absolute_configfile)

...

多配置文件

import os
from configHandle import Config

configfile = os.getenv("PROJECT_CONFIG_FILE", default='config_and_data_files/config.yaml')
pgm_configfile = os.getenv("PROJECT_PGM_CONFIG_FILE", default='config_and_data_files/pgm_config.yaml')
absolute_configfiles = map(lambda x:os.path.join(os.getcwd(), x), (configfile, pgm_configfile))

config = Config(absolute_configfiles)

...

比命令行参数版本更简洁了,另外还能避免在终端留下信息,其实环境变量是更优的选择。不过我更喜欢传入参数,因为形式简单,显式。


Linux 上设置环境变量:

export PROJECT_CONFIG_FILE=config.yaml

或者在 Windows 中:

set PROJECT_CONFIG_FILE=config.yaml

其实还支持 .env 文件导入环境变量的方式,不过这种写入文件,那还不如直接用 .yaml

.env 文件的内容可能如下所示:

DB_POSTGRES_HOST=localhost
DB_POSTGRES_PORT=5432
APP_WEBSERVER_PORT=80

要加载此 .env 文件,可以使用 python-dotenv 库。该库加载文件但不会覆盖现有的环境变量。

from dotenv import load_dotenv
import os

load_dotenv(".env")

db_postgres_host = os.getenv("DB_POSTGRES_HOST")
db_postgres_port = os.getenv("DB_POSTGRES_PORT")
app_webserver_port = os.getenv("APP_WEBSERVER_PORT")

这些代码可能有错误,因为我是从自己项目中剪切出来的,还修正了一些过去不合适的地方,没有经过使用,若有误敬请告知。

原文链接: https://yanh.tech/2024/04/best-practice-for-configuration-and-data-rw-in-small-and-medium-python-projects/

版权声明:本博客所有文章除特別声明外,均为 AhFei 原创,采用 CC BY-NC-SA 4.0 许可协议。转载请注明来源 技焉洲 (yanh.tech)

保持更新 ٩(•̤̀ᵕ•̤́๑)ᵒᵏᵎᵎᵎᵎ 清晰恒益的实用技能,欢迎使用 RSS 订阅,如果能留言互动就更好了。

可在 Telegram 群组 https://t.me/vfly2 交流依文章步骤遇到的问题。

评论

  1. Bach239
    2 月前
    2024-5-05 18:26:29

    若使用 Dynaconf 組織配置是否會更爲簡潔?

    • Bach239
      2 月前
      2024-5-06 0:00:11

      好像确实不错,功能上能覆盖我文中的配置管理需求。

      不过对于中小型项目,为了配置引入一个包也许有点大题小做。另外自己写配置类也更灵活吧,都是这样。

  2. PinHsin
    2 月前
    2024-5-06 16:58:40

    配置文件中的参数,手动提取出来

    self.is_production = configs[‘is_production’]
    大佬这一步中配置文件的每项都需要手动填写吗

    • PinHsin
      2 月前
      2024-5-06 17:19:53

      目前是这样的,属于重复性劳动。昨天看了 Dynaconf 给了启发,它会按照配置项原本结构自动生成属性,以后探索一下

发送评论 编辑评论


				
|´・ω・)ノ
ヾ(≧∇≦*)ゝ
(☆ω☆)
(╯‵□′)╯︵┴─┴
 ̄﹃ ̄
(/ω\)
∠( ᐛ 」∠)_
(๑•̀ㅁ•́ฅ)
→_→
୧(๑•̀⌄•́๑)૭
٩(ˊᗜˋ*)و
(ノ°ο°)ノ
(´இ皿இ`)
⌇●﹏●⌇
(ฅ´ω`ฅ)
(╯°A°)╯︵○○○
φ( ̄∇ ̄o)
ヾ(´・ ・`。)ノ"
( ง ᵒ̌皿ᵒ̌)ง⁼³₌₃
(ó﹏ò。)
Σ(っ °Д °;)っ
( ,,´・ω・)ノ"(´っω・`。)
╮(╯▽╰)╭
o(*////▽////*)q
>﹏<
( ๑´•ω•) "(ㆆᴗㆆ)
😂
😀
😅
😊
🙂
🙃
😌
😍
😘
😜
😝
😏
😒
🙄
😳
😡
😔
😫
😱
😭
💩
👻
🙌
🖕
👍
👫
👬
👭
🌚
🌝
🙈
💊
😶
🙏
🍦
🍉
😣
Source: github.com/k4yt3x/flowerhd
颜文字
Emoji
小恐龙
花!
上一篇
下一篇