我来介绍几种同步预发环境灰度数据的Python脚本方案：
基于数据库同步
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
预发环境灰度数据同步脚本
支持多种数据库类型
"""
import pymysql
import psycopg2
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class DataSyncManager:
"""数据同步管理器"""
def __init__(self, config_file: str = 'sync_config.json'):
self.config = self.load_config(config_file)
self.source_conn = None
self.target_conn = None
def load_config(self, config_file: str) -> Dict:
"""加载配置文件"""
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
def connect_databases(self):
"""连接源和目标数据库"""
try:
# 源数据库（灰度环境）
if self.config['source']['type'] == 'mysql':
self.source_conn = pymysql.connect(
host=self.config['source']['host'],
port=self.config['source']['port'],
user=self.config['source']['user'],
password=self.config['source']['password'],
database=self.config['source']['database'],
charset='utf8mb4'
)
elif self.config['source']['type'] == 'postgresql':
self.source_conn = psycopg2.connect(
host=self.config['source']['host'],
port=self.config['source']['port'],
user=self.config['source']['user'],
password=self.config['source']['password'],
dbname=self.config['source']['database']
)
# 目标数据库（预发环境）
if self.config['target']['type'] == 'mysql':
self.target_conn = pymysql.connect(
host=self.config['target']['host'],
port=self.config['target']['port'],
user=self.config['target']['user'],
password=self.config['target']['password'],
database=self.config['target']['database'],
charset='utf8mb4'
)
logger.info("数据库连接成功")
except Exception as e:
logger.error(f"数据库连接失败: {e}")
raise
def get_grayscale_data(self, table_name: str, conditions: Dict = None) -> List[Dict]:
"""获取灰度数据"""
try:
cursor = self.source_conn.cursor(pymysql.cursors.DictCursor)
# 构建查询SQL
sql = f"SELECT * FROM {table_name}"
if conditions:
where_clause = " AND ".join([f"{k}=%s" for k in conditions.keys()])
sql += f" WHERE {where_clause}"
cursor.execute(sql, list(conditions.values()))
else:
cursor.execute(sql)
data = cursor.fetchall()
logger.info(f"从灰度环境获取 {len(data)} 条数据")
return data
except Exception as e:
logger.error(f"获取数据失败: {e}")
raise
finally:
cursor.close()
def sync_data(self, table_name: str, data: List[Dict], sync_mode: str = 'upsert'):
"""同步数据到预发环境
Args:
table_name: 表名
data: 要同步的数据
sync_mode: 同步模式 ('insert', 'upsert', 'replace')
"""
try:
cursor = self.target_conn.cursor()
if not data:
logger.warning("没有数据需要同步")
return
# 获取表结构（用于构建SQL）
columns = list(data[0].keys())
placeholders = ", ".join(["%s"] * len(columns))
columns_str = ", ".join(columns)
if sync_mode == 'upsert':
# UPSERT模式（MySQL使用ON DUPLICATE KEY UPDATE）
update_parts = []
for col in columns:
if col != 'id': # 假设id是主键
update_parts.append(f"{col}=VALUES({col})")
update_str = ", ".join(update_parts)
sql = f"""
INSERT INTO {table_name} ({columns_str})
VALUES ({placeholders})
ON DUPLICATE KEY UPDATE {update_str}
"""
elif sync_mode == 'replace':
# REPLACE模式
sql = f"REPLACE INTO {table_name} ({columns_str}) VALUES ({placeholders})"
else:
# 简单的INSERT模式
sql = f"INSERT IGNORE INTO {table_name} ({columns_str}) VALUES ({placeholders})"
# 批量插入
batch_size = 1000
total_synced = 0
for i in range(0, len(data), batch_size):
batch = data[i:i+batch_size]
values = [tuple(item[col] for col in columns) for item in batch]
cursor.executemany(sql, values)
self.target_conn.commit()
total_synced += len(batch)
logger.info(f"同步进度: {total_synced}/{len(data)}")
logger.info(f"数据同步完成，共同步 {total_synced} 条记录")
except Exception as e:
self.target_conn.rollback()
logger.error(f"数据同步失败: {e}")
raise
finally:
cursor.close()
def close_connections(self):
"""关闭数据库连接"""
if self.source_conn:
self.source_conn.close()
if self.target_conn:
self.target_conn.close()
logger.info("数据库连接已关闭")
def sync_by_time_range(self, table_name: str, time_field: str,
start_time: datetime, end_time: datetime):
"""按时间范围同步数据"""
conditions = {
time_field: (start_time, end_time)
}
data = self.get_grayscale_data(table_name, conditions)
if data:
self.sync_data(table_name, data)
# 配置文件示例 (sync_config.json)
"""
{
"source": {
"type": "mysql",
"host": "gray-db.example.com",
"port": 3306,
"user": "gray_user",
"password": "your_password",
"database": "gray_db"
},
"target": {
"type": "mysql",
"host": "staging-db.example.com",
"port": 3306,
"user": "staging_user",
"password": "your_password",
"database": "staging_db"
},
"sync_settings": {
"mode": "upsert",
"batch_size": 1000,
"retry_times": 3
}
}
"""
# 使用示例
if __name__ == "__main__":
sync_manager = DataSyncManager('sync_config.json')
try:
# 连接数据库
sync_manager.connect_databases()
# 示例1：同步所有灰度用户数据
user_conditions = {"is_grayscale": 1, "status": 1}
user_data = sync_manager.get_grayscale_data("users", user_conditions)
sync_manager.sync_data("users", user_data, sync_mode='upsert')
# 示例2：按时间范围同步
start_time = datetime.now() - timedelta(hours=1)
end_time = datetime.now()
sync_manager.sync_by_time_range("orders", "created_at", start_time, end_time)
finally:
sync_manager.close_connections()
基于Redis缓存同步
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Redis灰度数据同步脚本
"""
import redis
import json
import logging
from typing import Dict, Any, Optional
logger = logging.getLogger(__name__)
class RedisDataSyncer:
"""Redis灰度数据同步"""
def __init__(self, config: Dict):
self.gray_redis = redis.Redis(
host=config['gray']['host'],
port=config['gray']['port'],
password=config['gray'].get('password'),
decode_responses=True
)
self.staging_redis = redis.Redis(
host=config['staging']['host'],
port=config['staging']['port'],
password=config['staging'].get('password'),
decode_responses=True
)
def sync_string_data(self, key: str):
"""同步String类型数据"""
try:
value = self.gray_redis.get(key)
if value is not None:
self.staging_redis.set(key, value)
logger.info(f"同步String数据成功: {key}")
else:
logger.warning(f"源Redis中不存在key: {key}")
except Exception as e:
logger.error(f"同步String数据失败: {e}")
def sync_hash_data(self, key: str, fields: Optional[str] = None):
"""同步Hash类型数据"""
try:
if fields:
data = self.gray_redis.hget(key, fields)
if data:
self.staging_redis.hset(key, fields, data)
else:
data = self.gray_redis.hgetall(key)
if data:
self.staging_redis.hset(key, mapping=data)
logger.info(f"同步Hash数据成功: {key}")
except Exception as e:
logger.error(f"同步Hash数据失败: {e}")
def sync_list_data(self, key: str):
"""同步List类型数据"""
try:
data = self.gray_redis.lrange(key, 0, -1)
if data:
# 清空目标key并重新添加
self.staging_redis.delete(key)
self.staging_redis.rpush(key, *data)
logger.info(f"同步List数据成功: {key}, 数量: {len(data)}")
except Exception as e:
logger.error(f"同步List数据失败: {e}")
def sync_set_data(self, key: str):
"""同步Set类型数据"""
try:
members = self.gray_redis.smembers(key)
if members:
self.staging_redis.delete(key)
self.staging_redis.sadd(key, *members)
logger.info(f"同步Set数据成功: {key}, 数量: {len(members)}")
except Exception as e:
logger.error(f"同步Set数据失败: {e}")
def sync_by_pattern(self, pattern: str = "grayscale:*"):
"""按模式同步所有灰度数据"""
cursor = 0
while True:
cursor, keys = self.gray_redis.scan(cursor, match=pattern, count=100)
for key in keys:
key_type = self.gray_redis.type(key)
if key_type == 'string':
self.sync_string_data(key)
elif key_type == 'hash':
self.sync_hash_data(key)
elif key_type == 'list':
self.sync_list_data(key)
elif key_type == 'set':
self.sync_set_data(key)
if cursor == 0:
break
def sync_specific_keys(self, keys: list):
"""同步指定的key列表"""
for key in keys:
key_type = self.gray_redis.type(key)
if key_type == 'string':
self.sync_string_data(key)
elif key_type == 'hash':
self.sync_hash_data(key)
elif key_type == 'list':
self.sync_list_data(key)
elif key_type == 'set':
self.sync_set_data(key)
else:
logger.warning(f"不支持的数据类型: {key_type} for key: {key}")
# 使用示例
if __name__ == "__main__":
config = {
'gray': {
'host': 'gray-redis.example.com',
'port': 6379,
'password': 'gray_pass'
},
'staging': {
'host': 'staging-redis.example.com',
'port': 6379,
'password': 'staging_pass'
}
}
syncer = RedisDataSyncer(config)
# 同步所有灰度相关数据
syncer.sync_by_pattern("grayscale:*")
# 同步特定key
specific_keys = [
"grayscale:user:config:123",
"grayscale:feature:flag:new_ui"
]
syncer.sync_specific_keys(specific_keys)
基于API调用同步
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
基于API调用的灰度数据同步
"""
import requests
import json
import hashlib
import time
from typing import Dict, List, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
class APIDataSyncer:
"""API数据同步器"""
def __init__(self, config: Dict):
self.source_api = config['source_api']
self.target_api = config['target_api']
self.api_key = config.get('api_key', '')
self.api_secret = config.get('api_secret', '')
self.session = requests.Session()
def generate_signature(self, params: Dict) -> str:
"""生成API签名"""
sorted_params = sorted(params.items())
param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
sign_str = f"{param_str}{self.api_secret}"
return hashlib.md5(sign_str.encode()).hexdigest()
def call_api(self, url: str, method: str = 'GET',
params: Dict = None, data: Dict = None) -> Dict:
"""调用API接口"""
headers = {
'Content-Type': 'application/json',
'X-API-Key': self.api_key
}
if params:
params['timestamp'] = int(time.time())
params['sign'] = self.generate_signature(params)
try:
if method == 'GET':
response = self.session.get(url, params=params, headers=headers)
elif method == 'POST':
response = self.session.post(url, json=data, params=params,
headers=headers)
elif method == 'PUT':
response = self.session.put(url, json=data, params=params,
headers=headers)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"API调用失败: {e}")
return {'error': str(e)}
def get_grayscale_data(self, endpoint: str,
params: Dict = None) -> List[Dict]:
"""从源API获取灰度数据"""
url = f"{self.source_api}/{endpoint}"
response = self.call_api(url, params=params)
if 'error' in response:
return []
return response.get('data', [])
def post_data_to_target(self, endpoint: str, data: Dict) -> bool:
"""将数据POST到目标API"""
url = f"{self.target_api}/{endpoint}"
response = self.call_api(url, method='POST', data=data)
if 'error' not in response:
return True
return False
def sync_grayscale_users(self):
"""同步灰度用户数据"""
# 获取灰度用户列表
users = self.get_grayscale_data('/api/users/grayscale')
success_count = 0
for user in users:
# 同步到目标环境
if self.post_data_to_target('/api/users/sync', user):
success_count += 1
print(f"同步用户 {user['id']} 成功")
else:
print(f"同步用户 {user['id']} 失败")
print(f"同步完成: 成功 {success_count}/{len(users)}")
def sync_with_threads(self, endpoint: str, max_workers: int = 10):
"""多线程同步数据"""
data_list = self.get_grayscale_data(endpoint)
def sync_item(item):
return self.post_data_to_target(endpoint, item)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(sync_item, item): item for item in data_list}
for future in as_completed(futures):
item = futures[future]
try:
result = future.result()
if result:
print(f"同步数据成功: {item.get('id')}")
except Exception as e:
print(f"同步数据失败: {item.get('id')}, 错误: {e}")
# 使用示例
if __name__ == "__main__":
config = {
"source_api": "https://gray-api.example.com",
"target_api": "https://staging-api.example.com",
"api_key": "your_api_key",
"api_secret": "your_api_secret"
}
syncer = APIDataSyncer(config)
# 同步灰度用户
syncer.sync_grayscale_users()
# 或使用多线程同步
syncer.sync_with_threads('/api/orders/grayscale', max_workers=5)
配置文件模板
创建一个 sync_config.json 配置文件：
{
"source": {
"type": "mysql",
"host": "gray-db.example.com",
"port": 3306,
"user": "gray_user",
"password": "encrypted_password",
"database": "gray_db",
"charset": "utf8mb4"
},
"target": {
"type": "mysql",
"host": "staging-db.example.com",
"port": 3306,
"user": "staging_user",
"password": "encrypted_password",
"database": "staging_db",
"charset": "utf8mb4"
},
"sync_settings": {
"mode": "upsert",
"batch_size": 1000,
"max_retries": 3,
"retry_delay": 5,
"timeout": 300,
"exclude_tables": ["logs", "audit_trails"],
"include_tables": ["users", "orders", "products"],
"filter_conditions": {
"users": {"is_grayscale": 1, "status": "active"},
"orders": {"is_grayscale": 1}
},
"field_mappings": {
"users": {
"created_at": "created_at",
"updated_at": "updated_at"
}
},
"before_sync_hook": "backup_data",
"after_sync_hook": "verify_sync",
"schedule": {
"cron": "0 */6 * * *",
"timezone": "Asia/Shanghai"
}
},
"notifications": {
"email": {
"enabled": true,
"recipients": ["team@example.com"],
"on_success": false,
"on_failure": true
},
"webhook": {
"enabled": true,
"url": "https://hooks.example.com/sync",
"events": ["failure", "completed"]
},
"slack": {
"enabled": true,
"webhook_url": "https://hooks.slack.com/services/xxx",
"channel": "#data-sync"
}
},
"monitoring": {
"enabled": true,
"metrics_endpoint": "http://metrics.example.com/push",
"log_level": "INFO",
"audit_log": "/var/log/data_sync.log"
},
"security": {
"encrypt_passwords": true,
"use_vault": false,
"vault_address": "http://vault.example.com:8200",
"allowed_ips": ["10.0.0.0/8", "172.16.0.0/12"],
"sync_window": "09:00-18:00"
},
"error_handling": {
"max_errors": 100,
"error_threshold": 10,
"action_on_threshold": "pause",
"dead_letter_queue": true
}
}
使用说明
-
安装依赖：
pip install pymysql psycopg2-binary redis requests python-dotenv
-
配置环境变量：
# .env file SOURCE_DB_HOST=gray-db.example.com SOURCE_DB_USER=gray_user SOURCE_DB_PASSWORD=secret_password TARGET_DB_HOST=staging-db.example.com TARGET_DB_USER=staging_user TARGET_DB_PASSWORD=secret_password
-
运行脚本：
# 直接运行 python data_sync.py
使用crontab定时执行
0 /6 /usr/bin/python3 /path/to/data_sync.py >> /var/log/sync.log 2>&1
4. **常见问题处理**：
- **连接超时**：增加连接超时时间，使用连接池
- **数据不一致**：实现数据校验和回滚机制
- **性能问题**：使用批量操作，多线程并发
- **数据量过大**：分页处理，增量同步
这些方案可以根据你的具体场景选择使用或组合使用。
