我来介绍几种同步预发环境灰度数据的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. **常见问题处理**：
- **连接超时**：增加连接超时时间，使用连接池
- **数据不一致**：实现数据校验和回滚机制
- **性能问题**：使用批量操作，多线程并发
- **数据量过大**：分页处理，增量同步
这些方案可以根据你的具体场景选择使用或组合使用。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
