如何为Python程序配置互斥输入源的配置文件结构？

一、结构化配置设计（基于dataclass） #

针对互斥数据源场景，我们可以用typing.Union标记互斥的配置类，同时添加类型标识字段区分数据源类型，既保证配置结构化，又能清晰识别要使用的数据源。

示例代码 #

from dataclasses import dataclass
from typing import Union, Optional

# 通用基础配置
@dataclass
class BaseConfig:
    output_path: str
    chunk_size: int = 1000

# CSV专属配置
@dataclass
class CSVConfig:
    type: str = "csv"
    file_path: str
    delimiter: str = ","
    encoding: str = "utf-8"

# 数据库专属配置
@dataclass
class DBConfig:
    type: str = "db"
    host: str
    port: int
    database: str
    username: str
    password: str
    query: str

# 总配置：基础配置 + 互斥数据源配置
@dataclass
class AppConfig:
    base: BaseConfig
    source: Union[CSVConfig, DBConfig]

配置文件示例（JSON格式，标准库可直接解析） #

CSV版本：

{
    "base": {
        "output_path": "./processed_data",
        "chunk_size": 2000
    },
    "source": {
        "type": "csv",
        "file_path": "./input.csv",
        "delimiter": ";"
    }
}

数据库版本：

{
    "base": {
        "output_path": "./processed_data",
        "chunk_size": 2000
    },
    "source": {
        "type": "db",
        "host": "localhost",
        "port": 5432,
        "database": "mydb",
        "username": "admin",
        "password": "123456",
        "query": "SELECT * FROM raw_data"
    }
}

配置解析逻辑 #

import json
from typing import Type

def parse_config(config_path: str) -> AppConfig:
    with open(config_path, "r") as f:
        config_dict = json.load(f)
    
    base_config = BaseConfig(**config_dict["base"])
    source_type = config_dict["source"]["type"]
    source_cls: Type[Union[CSVConfig, DBConfig]] = CSVConfig if source_type == "csv" else DBConfig
    source_config = source_cls(**config_dict["source"])
    
    return AppConfig(base=base_config, source=source_config)

二、策略模式封装数据获取逻辑 #

用Protocol定义统一的数据获取接口，分别实现CSV和数据库的数据源类，避免大量if判断，直接根据配置类型实例化对应策略。

接口与实现代码 #

from typing import Protocol
import pandas as pd
import sqlite3  # 以SQLite为例，其他数据库可替换连接逻辑

class DataSource(Protocol):
    def fetch_data(self, chunk_size: int) -> pd.DataFrame:
        """统一数据获取接口"""
        ...

class CSVDataSource:
    def __init__(self, config: CSVConfig):
        self.config = config
    
    def fetch_data(self, chunk_size: int):
        return pd.read_csv(
            self.config.file_path,
            delimiter=self.config.delimiter,
            encoding=self.config.encoding,
            chunksize=chunk_size
        )

class DBDataSource:
    def __init__(self, config: DBConfig):
        self.config = config
    
    def fetch_data(self, chunk_size: int):
        # 数据库分批查询逻辑，可根据数据库类型调整
        conn = sqlite3.connect(f"sqlite:///{self.config.host}:{self.config.port}/{self.config.database}")
        offset = 0
        while True:
            query = f"{self.config.query} LIMIT {chunk_size} OFFSET {offset}"
            df = pd.read_sql(query, conn)
            if df.empty:
                break
            yield df
            offset += chunk_size
        conn.close()

三、消除显式判断的优化方案 #

维护类型映射字典，将数据源类型字符串与配置类、策略类绑定，新增数据源时只需扩展映射，无需修改核心逻辑。

映射与主逻辑代码 #

# 类型映射：key为source.type值，value为(配置类, 策略类)
SOURCE_MAP = {
    "csv": (CSVConfig, CSVDataSource),
    "db": (DBConfig, DBDataSource)
}

# 优化配置解析函数
def parse_config(config_path: str) -> AppConfig:
    with open(config_path, "r") as f:
        config_dict = json.load(f)
    
    base_config = BaseConfig(**config_dict["base"])
    source_type = config_dict["source"]["type"]
    source_cls, _ = SOURCE_MAP[source_type]
    source_config = source_cls(**config_dict["source"])
    
    return AppConfig(base=base_config, source=source_config)

# 主逻辑
def main(config: AppConfig):
    source_type = config.source.type
    _, data_source_cls = SOURCE_MAP[source_type]
    data_source = data_source_cls(config.source)
    
    # 统一数据处理流程
    for chunk in data_source.fetch_data(config.base.chunk_size):
        print(f"Processing chunk with {len(chunk)} rows")
        chunk.to_csv(f"{config.base.output_path}/chunk_{offset}.csv", index=False)

if __name__ == "__main__":
    config = parse_config("config.json")
    main(config)

四、替代配置方案：Python文件作为配置 #

如果需要更强的类型校验和可读性，可直接用Python文件定义配置，无需额外解析：

示例config.py #

from dataclasses import dataclass
from typing import Union

@dataclass
class BaseConfig:
    output_path: str
    chunk_size: int = 1000

@dataclass
class CSVConfig:
    type: str = "csv"
    file_path: str
    delimiter: str = ","
    encoding: str = "utf-8"

@dataclass
class DBConfig:
    type: str = "db"
    host: str
    port: int
    database: str
    username: str
    password: str
    query: str

@dataclass
class AppConfig:
    base: BaseConfig
    source: Union[CSVConfig, DBConfig]

# 具体配置实例
config = AppConfig(
    base=BaseConfig(
        output_path="./processed_data",
        chunk_size=2000
    ),
    source=CSVConfig(
        file_path="./input.csv",
        delimiter=";"
    )
)

主程序导入 #

from config import config

if __name__ == "__main__":
    main(config)

内容的提问来源于stack exchange，提问作者charlieshades