Python脚本如何按需同步业务模块数据

wen python案例 2026-06-25 265

本文目录导读：

基于配置驱动的同步方案
带依赖管理的同步方案
带缓存和断点续传的同步方案
命令行调用示例
使用建议

我来为你介绍几种按需同步业务模块数据的Python实现方案：

基于配置驱动的同步方案

import json
import time
from typing import Dict, List, Optional
from datetime import datetime
import logging
class BusinessDataSync:
    """业务数据同步器"""
    def __init__(self, config_file: str = 'sync_config.json'):
        self.logger = logging.getLogger(__name__)
        self.config = self.load_config(config_file)
        self.sync_history = []
    def load_config(self, config_file: str) -> Dict:
        """加载同步配置"""
        with open(config_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    def sync_module(self, module_name: str, sync_type: str = 'incremental'):
        """同步指定模块
        Args:
            module_name: 模块名称
            sync_type: 同步类型 (full/incremental)
        """
        if module_name not in self.config['modules']:
            self.logger.error(f"未找到模块: {module_name}")
            return False
        module_config = self.config['modules'][module_name]
        # 获取数据源
        source_data = self._get_source_data(module_config, sync_type)
        # 数据转换
        transformed_data = self._transform_data(source_data, module_config)
        # 写入目标
        success = self._write_to_target(transformed_data, module_config)
        # 记录同步历史
        self._record_sync_history(module_name, sync_type, success)
        return success
    def _get_source_data(self, config: Dict, sync_type: str) -> List[Dict]:
        """获取源数据"""
        source_type = config.get('source_type', 'api')
        if source_type == 'api':
            return self._fetch_from_api(config['source_config'])
        elif source_type == 'database':
            return self._fetch_from_db(config['source_config'])
        elif source_type == 'file':
            return self._fetch_from_file(config['source_config'])
        else:
            raise ValueError(f"不支持的数据源类型: {source_type}")
    def _fetch_from_api(self, config: Dict) -> List[Dict]:
        """从API获取数据"""
        import requests
        params = {
            'page_size': config.get('page_size', 100),
            'page': 1
        }
        # 增量同步时添加时间过滤
        if config.get('last_sync_time'):
            params['updated_after'] = config['last_sync_time']
        all_data = []
        while True:
            response = requests.get(config['url'], params=params)
            data = response.json()
            all_data.extend(data.get('items', []))
            if not data.get('has_more'):
                break
            params['page'] += 1
        return all_data
    def _transform_data(self, data: List[Dict], config: Dict) -> List[Dict]:
        """数据转换"""
        mapping = config.get('field_mapping', {})
        transformed = []
        for item in data:
            new_item = {}
            for source_field, target_field in mapping.items():
                new_item[target_field] = item.get(source_field)
            transformed.append(new_item)
        return transformed
    def _write_to_target(self, data: List[Dict], config: Dict) -> bool:
        """写入目标系统"""
        target_type = config.get('target_type', 'database')
        if target_type == 'database':
            return self._write_to_db(data, config['target_config'])
        elif target_type == 'file':
            return self._write_to_file(data, config['target_config'])
        elif target_type == 'redis':
            return self._write_to_redis(data, config['target_config'])
        else:
            raise ValueError(f"不支持的目标类型: {target_type}")
    def _record_sync_history(self, module: str, sync_type: str, success: bool):
        """记录同步历史"""
        self.sync_history.append({
            'module': module,
            'sync_type': sync_type,
            'timestamp': datetime.now().isoformat(),
            'success': success
        })
# 配置文件示例 (sync_config.json)
"""
{
    "modules": {
        "user": {
            "source_type": "api",
            "source_config": {
                "url": "https://api.example.com/users",
                "page_size": 100
            },
            "target_type": "database",
            "target_config": {
                "host": "localhost",
                "database": "business_db",
                "table": "users"
            },
            "field_mapping": {
                "id": "user_id",
                "name": "username",
                "email": "email"
            }
        },
        "order": {
            "source_type": "api",
            "source_config": {
                "url": "https://api.example.com/orders",
                "page_size": 50
            },
            "target_type": "file",
            "target_config": {
                "path": "/data/orders/",
                "format": "json"
            },
            "field_mapping": {
                "order_id": "id",
                "total_amount": "amount"
            }
        }
    }
}
"""
# 使用示例
sync = BusinessDataSync('sync_config.json')
# 按需同步用户模块
sync.sync_module('user', 'incremental')
# 全量同步订单模块
sync.sync_module('order', 'full')

带依赖管理的同步方案

from typing import Dict, List, Set
from enum import Enum
class SyncStatus(Enum):
    PENDING = 'pending'
    SYNCING = 'syncing'
    COMPLETED = 'completed'
    FAILED = 'failed'
class DependencySyncManager:
    """依赖同步管理器"""
    def __init__(self):
        self.dependencies = {
            'user': [],           # 用户模块无依赖
            'order': ['user'],    # 订单模块依赖用户
            'product': ['category'],  # 产品模块依赖分类
            'category': [],       # 分类模块无依赖
            'inventory': ['product', 'warehouse'],  # 库存依赖产品和仓库
            'warehouse': []
        }
        self.module_status = {}
        self.sync_handlers = {}
    def register_handler(self, module: str, handler_func):
        """注册同步处理器"""
        self.sync_handlers[module] = handler_func
    def get_sync_order(self, modules: List[str]) -> List[str]:
        """获取同步顺序（拓扑排序）"""
        from collections import deque
        # 计算入度
        in_degree = {m: 0 for m in modules}
        graph = {m: [] for m in modules}
        for module in modules:
            for dep in self.dependencies.get(module, []):
                if dep in modules:
                    in_degree[module] = in_degree.get(module, 0) + 1
                    graph[dep].append(module)
        # BFS拓扑排序
        queue = deque([m for m in modules if in_degree[m] == 0])
        sorted_modules = []
        while queue:
            module = queue.popleft()
            sorted_modules.append(module)
            for next_module in graph[module]:
                in_degree[next_module] -= 1
                if in_degree[next_module] == 0:
                    queue.append(next_module)
        # 检查是否有循环依赖
        if len(sorted_modules) != len(modules):
            raise ValueError("检测到循环依赖关系")
        return sorted_modules
    def sync_modules_with_dependencies(self, modules: List[str]):
        """同步指定模块及其依赖"""
        # 获取所有需要同步的模块（包括依赖）
        modules_to_sync = set()
        for module in modules:
            modules_to_sync.add(module)
            modules_to_sync.update(self._get_all_dependencies(module))
        # 获取同步顺序
        sync_order = self.get_sync_order(list(modules_to_sync))
        results = {}
        print(f"同步顺序: {sync_order}")
        for module in sync_order:
            print(f"正在同步模块: {module}")
            self.module_status[module] = SyncStatus.SYNCING
            try:
                if module in self.sync_handlers:
                    result = self.sync_handlers[module]()
                    results[module] = result
                    self.module_status[module] = SyncStatus.COMPLETED
                    print(f"模块 {module} 同步完成")
                else:
                    raise ValueError(f"未找到模块 {module} 的同步处理器")
            except Exception as e:
                self.module_status[module] = SyncStatus.FAILED
                print(f"模块 {module} 同步失败: {e}")
                results[module] = None
        return results
    def _get_all_dependencies(self, module: str) -> Set[str]:
        """获取模块的所有依赖（递归）"""
        dependencies = set()
        for dep in self.dependencies.get(module, []):
            if dep not in dependencies:
                dependencies.add(dep)
                dependencies.update(self._get_all_dependencies(dep))
        return dependencies
# 使用示例
def sync_user_module():
    print("执行用户数据同步...")
    return {"synced_records": 100}
def sync_order_module():
    print("执行订单数据同步...")
    return {"synced_records": 200}
def sync_product_module():
    print("执行产品数据同步...")
    return {"synced_records": 50}
# 初始化管理器
manager = DependencySyncManager()
manager.register_handler('user', sync_user_module)
manager.register_handler('order', sync_order_module)
manager.register_handler('product', sync_product_module)
manager.register_handler('category', lambda: {"synced_records": 30})
# 按需同步（自动处理依赖）
results = manager.sync_modules_with_dependencies(['order', 'product'])
print(f"同步结果: {results}")

带缓存和断点续传的同步方案

import pickle
import hashlib
from datetime import datetime, timedelta
class CacheSyncManager:
    """带缓存的同步管理器"""
    def __init__(self, cache_dir: str = './sync_cache'):
        import os
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
        self.cache_ttl = {
            'user': timedelta(hours=1),
            'order': timedelta(minutes=30),
            'product': timedelta(days=1)
        }
    def get_cache_key(self, module: str, params: Dict = None) -> str:
        """生成缓存键"""
        base_key = f"sync_{module}"
        if params:
            param_str = json.dumps(params, sort_keys=True)
            base_key = f"{base_key}_{hashlib.md5(param_str.encode()).hexdigest()[:8]}"
        return base_key
    def get_cached_data(self, module: str, params: Dict = None) -> Optional[List[Dict]]:
        """获取缓存数据"""
        cache_key = self.get_cache_key(module, params)
        cache_file = f"{self.cache_dir}/{cache_key}.pkl"
        try:
            with open(cache_file, 'rb') as f:
                cached_data = pickle.load(f)
            # 检查缓存有效期
            if datetime.now() - cached_data['timestamp'] < self.cache_ttl.get(module, timedelta(hours=1)):
                return cached_data['data']
        except (FileNotFoundError, pickle.UnpicklingError):
            pass
        return None
    def set_cached_data(self, module: str, data: List[Dict], params: Dict = None):
        """设置缓存数据"""
        cache_key = self.get_cache_key(module, params)
        cache_file = f"{self.cache_dir}/{cache_key}.pkl"
        cached_data = {
            'timestamp': datetime.now(),
            'data': data,
            'module': module
        }
        with open(cache_file, 'wb') as f:
            pickle.dump(cached_data, f)
    def sync_with_checkpoint(self, module: str, batch_size: int = 100):
        """带检查点的同步（支持断点续传）"""
        checkpoint_file = f"{self.cache_dir}/checkpoint_{module}.json"
        # 检查是否有检查点
        checkpoint = self._load_checkpoint(checkpoint_file)
        start_id = checkpoint.get('last_processed_id', 0) if checkpoint else 0
        print(f"从ID {start_id} 继续同步模块 {module}")
        processed_count = 0
        has_more = True
        while has_more:
            try:
                # 分批获取数据
                batch_data = self._fetch_batch(module, start_id, batch_size)
                if not batch_data:
                    has_more = False
                    break
                # 处理批次数据
                self._process_batch(module, batch_data)
                # 更新检查点
                last_id = batch_data[-1].get('id')
                self._save_checkpoint(checkpoint_file, {
                    'last_processed_id': last_id,
                    'processed_count': processed_count + len(batch_data),
                    'last_update': datetime.now().isoformat()
                })
                processed_count += len(batch_data)
                start_id = last_id
                print(f"已同步 {processed_count} 条记录")
            except Exception as e:
                print(f"同步中断于ID {start_id}: {e}")
                return {
                    'success': False,
                    'processed_count': processed_count,
                    'next_checkpoint': start_id
                }
        # 同步完成，删除检查点
        import os
        os.remove(checkpoint_file)
        return {
            'success': True,
            'processed_count': processed_count
        }
    def _load_checkpoint(self, filepath: str) -> Optional[Dict]:
        """加载检查点"""
        try:
            with open(filepath, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return None
    def _save_checkpoint(self, filepath: str, checkpoint: Dict):
        """保存检查点"""
        with open(filepath, 'w') as f:
            json.dump(checkpoint, f)
    def _fetch_batch(self, module: str, start_id: int, batch_size: int) -> List[Dict]:
        """分批获取数据"""
        # 模拟数据库查询
        import random
        return [
            {'id': start_id + i, 'data': f'data_{i}'}
            for i in range(batch_size)
            if random.random() > 0.1  # 模拟数据终止
        ]
    def _process_batch(self, module: str, batch_data: List[Dict]):
        """处理批次数据"""
        # 实际应用中的业务处理
        pass
# 使用示例
cache_manager = CacheSyncManager()
# 检查缓存
cached_users = cache_manager.get_cached_data('user')
if cached_users:
    print("使用缓存数据")
else:
    print("缓存过期或不存在，执行同步")
    # 执行同步逻辑
    new_data = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    cache_manager.set_cached_data('user', new_data)
# 断点续传同步
result = cache_manager.sync_with_checkpoint('order', batch_size=50)
print(f"同步结果: {result}")

命令行调用示例

import argparse
import sys
def main():
    parser = argparse.ArgumentParser(description='业务数据同步工具')
    parser.add_argument('--modules', nargs='+', required=True, help='需要同步的模块')
    parser.add_argument('--sync-type', choices=['full', 'incremental'], default='incremental')
    parser.add_argument('--config', default='sync_config.json', help='配置文件路径')
    parser.add_argument('--parallel', action='store_true', help='是否并行同步')
    parser.add_argument('--force', action='store_true', help='强制重新同步')
    args = parser.parse_args()
    # 初始化同步器
    from concurrent.futures import ThreadPoolExecutor
    if args.parallel:
        # 并行同步
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = []
            for module in args.modules:
                future = executor.submit(
                    sync_module_wrapper,
                    module,
                    args.sync_type,
                    args.config,
                    args.force
                )
                futures.append(future)
            for future in futures:
                result = future.result()
                print(f"同步结果: {result}")
    else:
        # 串行同步
        for module in args.modules:
            result = sync_module_wrapper(module, args.sync_type, args.config, args.force)
            print(f"模块 {module} 同步 {'成功' if result else '失败'}")
def sync_module_wrapper(module, sync_type, config, force):
    """同步模块包装函数"""
    sync = BusinessDataSync(config)
    return sync.sync_module(module, sync_type)
if __name__ == '__main__':
    main()

使用命令行调用：

# 同步指定模块
python sync.py --modules user order product
# 全量同步
python sync.py --modules user --sync-type full
# 并行同步
python sync.py --modules user order product --parallel
# 强制重新同步
python sync.py --modules user --force