150 lines
4.7 KiB
Python
Executable File
150 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Clone Production Database to Dev Server
|
|
Helper script to clone the production database to a remote development server
|
|
"""
|
|
|
|
import argparse
|
|
import requests
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from backend.services.database_backup import DatabaseBackupService
|
|
|
|
|
|
def clone_to_dev(remote_url: str, snapshot_filename: str = None, auth_token: str = None):
|
|
"""Clone database to remote dev server"""
|
|
|
|
backup_service = DatabaseBackupService()
|
|
|
|
print(f"🔄 Cloning database to {remote_url}...")
|
|
|
|
try:
|
|
# If no snapshot specified, create a new one
|
|
if snapshot_filename:
|
|
print(f"📦 Using existing snapshot: {snapshot_filename}")
|
|
snapshot_path = backup_service.backups_dir / snapshot_filename
|
|
if not snapshot_path.exists():
|
|
print(f"❌ Error: Snapshot {snapshot_filename} not found")
|
|
return False
|
|
else:
|
|
print("📸 Creating new snapshot...")
|
|
snapshot_info = backup_service.create_snapshot(description="Clone to dev server")
|
|
snapshot_filename = snapshot_info["filename"]
|
|
snapshot_path = backup_service.backups_dir / snapshot_filename
|
|
print(f"✅ Snapshot created: {snapshot_filename} ({snapshot_info['size_mb']} MB)")
|
|
|
|
# Upload to remote server
|
|
print(f"📤 Uploading to {remote_url}...")
|
|
|
|
headers = {}
|
|
if auth_token:
|
|
headers["Authorization"] = f"Bearer {auth_token}"
|
|
|
|
with open(snapshot_path, 'rb') as f:
|
|
files = {'file': (snapshot_filename, f, 'application/x-sqlite3')}
|
|
|
|
response = requests.post(
|
|
f"{remote_url.rstrip('/')}/api/settings/database/upload-snapshot",
|
|
files=files,
|
|
headers=headers,
|
|
timeout=300
|
|
)
|
|
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
print(f"✅ Upload successful!")
|
|
print(f" Remote filename: {result['snapshot']['filename']}")
|
|
print(f" Size: {result['snapshot']['size_mb']} MB")
|
|
|
|
# Now restore on remote server
|
|
print("🔄 Restoring on remote server...")
|
|
restore_response = requests.post(
|
|
f"{remote_url.rstrip('/')}/api/settings/database/restore",
|
|
json={
|
|
"filename": result['snapshot']['filename'],
|
|
"create_backup": True
|
|
},
|
|
headers=headers,
|
|
timeout=60
|
|
)
|
|
|
|
restore_response.raise_for_status()
|
|
restore_result = restore_response.json()
|
|
|
|
print(f"✅ Database cloned successfully!")
|
|
print(f" Restored from: {restore_result['restored_from']}")
|
|
print(f" Remote backup created: {restore_result.get('backup_created', 'N/A')}")
|
|
|
|
return True
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"❌ Network error: {str(e)}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Error: {str(e)}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Clone production database to development server",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Clone current database to dev server
|
|
python clone_db_to_dev.py --url https://dev.example.com
|
|
|
|
# Clone using existing snapshot
|
|
python clone_db_to_dev.py --url https://dev.example.com --snapshot snapshot_20250101_120000.db
|
|
|
|
# Clone with authentication
|
|
python clone_db_to_dev.py --url https://dev.example.com --token YOUR_TOKEN
|
|
"""
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--url',
|
|
required=True,
|
|
help='Remote dev server URL (e.g., https://dev.example.com)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--snapshot',
|
|
help='Use existing snapshot instead of creating new one'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--token',
|
|
help='Authentication token for remote server'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
print("=" * 60)
|
|
print(" Database Cloning Tool - Production to Dev")
|
|
print("=" * 60)
|
|
print()
|
|
|
|
success = clone_to_dev(
|
|
remote_url=args.url,
|
|
snapshot_filename=args.snapshot,
|
|
auth_token=args.token
|
|
)
|
|
|
|
print()
|
|
if success:
|
|
print("🎉 Cloning completed successfully!")
|
|
sys.exit(0)
|
|
else:
|
|
print("💥 Cloning failed")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|