Python脚本如何动态适配业务数据变动

wen python案例 2026-06-25 29

本文目录导读：

配置文件驱动
数据库驱动适配
API接口动态获取
工厂模式动态切换
监听文件变化自动重载
综合示例：完整的动态适配系统
关键设计原则
最佳实践建议

我来详细介绍Python脚本动态适配业务数据变动的几种核心方法：

配置文件驱动

使用JSON/YAML配置文件

import json
import yaml
from typing import Any, Dict
class ConfigurableProcessor:
    def __init__(self, config_path: str):
        self.config = self._load_config(config_path)
    def _load_config(self, path: str) -> Dict:
        if path.endswith('.json'):
            with open(path, 'r') as f:
                return json.load(f)
        elif path.endswith('.yaml') or path.endswith('.yml'):
            with open(path, 'r') as f:
                return yaml.safe_load(f)
    def process_data(self, data: Dict) -> Any:
        # 动态使用配置中的规则
        if self.config.get('rules'):
            for rule in self.config['rules']:
                data = self._apply_rule(data, rule)
        return data
# 示例配置文件 content
"""
config.json:
{
    "rules": [
        {"type": "filter", "field": "age", "min": 18},
        {"type": "transform", "field": "name", "to_upper": true}
    ],
    "thresholds": {
        "price_discount": 0.2,
        "max_items": 100
    }
}
"""

数据库驱动适配

import sqlite3
import pandas as pd
from datetime import datetime
class DatabaseDrivenAdapter:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self._init_tables()
    def _init_tables(self):
        # 创建业务规则表
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS business_rules (
                id INTEGER PRIMARY KEY,
                rule_name TEXT,
                rule_type TEXT,
                rule_params TEXT,
                is_active BOOLEAN DEFAULT 1,
                created_at TIMESTAMP
            )
        ''')
        # 创建业务数据表
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS business_data (
                id INTEGER PRIMARY KEY,
                data_type TEXT,
                data_content TEXT,
                effective_date DATE
            )
        ''')
        self.conn.commit()
    def get_active_rules(self):
        cursor = self.conn.execute(
            "SELECT * FROM business_rules WHERE is_active = 1"
        )
        return [dict(row) for row in cursor.fetchall()]
    def get_current_business_data(self):
        today = datetime.now().date()
        cursor = self.conn.execute(
            "SELECT * FROM business_data WHERE effective_date <= ?",
            (today,)
        )
        return cursor.fetchall()

API接口动态获取

import requests
import json
from typing import Dict, List, Optional
class APIAdapter:
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {api_key}'} if api_key else {}
        self.cache = {}
    def fetch_business_rules(self, endpoint: str = '/rules') -> Dict:
        """从API获取最新的业务规则"""
        try:
            response = requests.get(
                f"{self.base_url}{endpoint}",
                headers=self.headers,
                timeout=5
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API请求失败: {e}")
            return {}
    def fetch_dynamic_config(self, params: Dict) -> Dict:
        """根据参数动态获取配置"""
        response = requests.post(
            f"{self.base_url}/config",
            json=params,
            headers=self.headers
        )
        return response.json()
# 使用示例
adapter = APIAdapter("https://api.example.com", "your-api-key")
rules = adapter.fetch_business_rules()

工厂模式动态切换

from abc import ABC, abstractmethod
class DataProcessor(ABC):
    @abstractmethod
    def process(self, data):
        pass
class ProcessorV1(DataProcessor):
    def process(self, data):
        return {"version": "v1", "result": data}
class ProcessorV2(DataProcessor):
    def process(self, data):
        return {"version": "v2", "result": data * 2}
class ProcessorV3(DataProcessor):
    def process(self, data):
        return {"version": "v3", "result": data ** 2}
class ProcessorFactory:
    processors = {
        "v1": ProcessorV1,
        "v2": ProcessorV2,
        "v3": ProcessorV3
    }
    @classmethod
    def get_processor(cls, version: str) -> DataProcessor:
        processor_class = cls.processors.get(version)
        if not processor_class:
            raise ValueError(f"Unknown version: {version}")
        return processor_class()
    @classmethod
    def register_processor(cls, version: str, processor_class):
        cls.processors[version] = processor_class

监听文件变化自动重载

import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class ConfigFileWatcher(FileSystemEventHandler):
    def __init__(self, config_file: str, callback):
        self.config_file = config_file
        self.callback = callback
        self.last_modified = 0
    def on_modified(self, event):
        if event.src_path == self.config_file:
            current_modified = os.path.getmtime(self.config_file)
            if current_modified > self.last_modified:
                self.last_modified = current_modified
                self.callback()
class DynamicConfigManager:
    def __init__(self, config_path: str):
        self.config_path = config_path
        self.config = {}
        self._start_watcher()
    def _load_config(self):
        with open(self.config_path, 'r') as f:
            self.config = json.load(f)
        print(f"配置已更新: {self.config}")
    def _start_watcher(self):
        event_handler = ConfigFileWatcher(
            self.config_path, 
            self._load_config
        )
        observer = Observer()
        observer.schedule(
            event_handler, 
            path=os.path.dirname(self.config_path),
            recursive=False
        )
        observer.start()

综合示例：完整的动态适配系统

import json
import os
import threading
import time
from typing import Any, Dict, Callable
class DynamicBusinessAdapter:
    def __init__(self, config_sources: list = None):
        self.adapters = []
        self.config = {}
        self.rules = {}
        self.data_sources = {}
        # 初始化配置源
        if config_sources:
            for source in config_sources:
                self.add_adapter(source)
    def add_adapter(self, adapter):
        """添加适配器"""
        self.adapters.append(adapter)
    def refresh_configuration(self):
        """刷新所有配置"""
        for adapter in self.adapters:
            if hasattr(adapter, 'get_config'):
                config = adapter.get_config()
                self.config.update(config)
    def process_business_data(self, data: Any) -> Any:
        """处理业务数据"""
        # 先更新配置
        self.refresh_configuration()
        # 根据配置处理数据
        processor = self._select_processor(data)
        return processor(data)
    def _select_processor(self, data: Any) -> Callable:
        """根据数据和配置选择处理器"""
        data_type = type(data).__name__
        if data_type in self.config.get('processors', {}):
            processor_name = self.config['processors'][data_type]
            return self._get_processor(processor_name)
        # 默认处理器
        return lambda x: x
    def _get_processor(self, name: str) -> Callable:
        """获取处理器函数"""
        processors = {
            'int': lambda x: x * 2,
            'str': lambda x: x.upper(),
            'list': lambda x: list(set(x)),
            'dict': lambda x: {k: v for k, v in x.items() if v}
        }
        return processors.get(name, lambda x: x)
    def schedule_config_refresh(self, interval: int = 60):
        """定时刷新配置"""
        def refresh_task():
            while True:
                self.refresh_configuration()
                time.sleep(interval)
        thread = threading.Thread(target=refresh_task, daemon=True)
        thread.start()
# 使用示例
def main():
    # 创建适配器
    adapter = DynamicBusinessAdapter([
        # 配置文件适配器
        ConfigFileAdapter('config.json'),
        # 数据库适配器
        DatabaseAdapter('business.db'),
        # API适配器
        APIAdapter('https://api.example.com')
    ])
    # 定时刷新配置（每30秒）
    adapter.schedule_config_refresh(30)
    # 处理业务数据
    business_data = {"name": "test", "price": 100, "quantity": 5}
    result = adapter.process_business_data(business_data)
    print(f"处理结果: {result}")
if __name__ == "__main__":
    main()

关键设计原则

松耦合：通过接口和抽象类隔离变化
可配置化：将业务参数外置到配置文件
热更新：支持运行时动态加载配置
版本控制：维护配置的历史版本
容错机制：配置加载失败时的默认行为

最佳实践建议

使用环境变量区分开发/生产配置
配置变更时记录变更日志
实现配置的灰度发布机制
提供配置的可视化界面管理
添加配置的监控告警功能

这些方法可以让你的Python脚本灵活适应业务数据的变动,而不需要频繁修改代码。