amo-server/scripts/backup_database.py
Maxim Snesarev 33d6bb7ebd Refactor AMO CRM Data Collection Service to use PostgreSQL
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings.
- Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure.
- Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic.
- Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks.
- Improved logging and error handling across the application to facilitate better monitoring and debugging.
- Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
2025-11-05 00:38:37 +03:00

180 lines
5.2 KiB
Python

#!/usr/bin/env python3
"""
Database Backup Script
Creates timestamped backups of the SQLite database and maintains
a rolling window of recent backups.
"""
import shutil
from datetime import datetime
from pathlib import Path
import argparse
def backup_database(db_path: str = "amo_data.db", keep_count: int = 7) -> None:
"""
Create a backup of the database.
Args:
db_path: Path to the database file
keep_count: Number of backups to keep (default: 7)
"""
db_file = Path(db_path)
if not db_file.exists():
print(f"✗ Database not found: {db_path}")
return
# Create backups directory
backup_dir = Path("backups")
backup_dir.mkdir(exist_ok=True)
# Create timestamped backup
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = backup_dir / f"{db_file.stem}_{timestamp}.db"
print(f"Creating backup: {backup_path}")
shutil.copy2(db_file, backup_path)
print(f"✓ Backup created successfully ({backup_path.stat().st_size / 1024 / 1024:.2f} MB)")
# Clean up old backups
backups = sorted(backup_dir.glob(f"{db_file.stem}_*.db"))
if len(backups) > keep_count:
print(f"\nCleaning up old backups (keeping last {keep_count}):")
for old_backup in backups[:-keep_count]:
print(f" Removing: {old_backup.name}")
old_backup.unlink()
print(f"✓ Removed {len(backups) - keep_count} old backup(s)")
# Show current backups
print(f"\nCurrent backups ({len(backups[-keep_count:])} total):")
for backup in backups[-keep_count:]:
size_mb = backup.stat().st_size / 1024 / 1024
mtime = datetime.fromtimestamp(backup.stat().st_mtime)
print(f" {backup.name:40} {size_mb:>8.2f} MB {mtime.strftime('%Y-%m-%d %H:%M:%S')}")
def list_backups(db_name: str = "amo_data") -> None:
"""List all available backups."""
backup_dir = Path("backups")
if not backup_dir.exists():
print("No backups directory found.")
return
backups = sorted(backup_dir.glob(f"{db_name}_*.db"))
if not backups:
print(f"No backups found for {db_name}.db")
return
print(f"Available backups for {db_name}.db:")
print("-" * 80)
for backup in backups:
size_mb = backup.stat().st_size / 1024 / 1024
mtime = datetime.fromtimestamp(backup.stat().st_mtime)
print(f"{backup.name:40} {size_mb:>8.2f} MB {mtime.strftime('%Y-%m-%d %H:%M:%S')}")
def restore_backup(backup_name: str, db_path: str = "amo_data.db") -> None:
"""
Restore database from a backup.
Args:
backup_name: Name of the backup file to restore
db_path: Path where to restore the database
"""
backup_dir = Path("backups")
backup_file = backup_dir / backup_name
if not backup_file.exists():
print(f"✗ Backup not found: {backup_file}")
return
db_file = Path(db_path)
# Backup current database before restoring
if db_file.exists():
current_backup = f"{db_file}.before_restore.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
print(f"Backing up current database to: {current_backup}")
shutil.copy2(db_file, current_backup)
# Restore from backup
print(f"Restoring from backup: {backup_name}")
shutil.copy2(backup_file, db_file)
print(f"✓ Database restored successfully")
def main() -> None:
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Backup and restore SQLite databases",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Create a backup
python backup_database.py backup
# List all backups
python backup_database.py list
# Restore from a specific backup
python backup_database.py restore amo_data_20241006_120000.db
# Create backup and keep last 14 backups
python backup_database.py backup --keep 14
"""
)
subparsers = parser.add_subparsers(dest="command", help="Command to execute")
# Backup command
backup_parser = subparsers.add_parser("backup", help="Create a database backup")
backup_parser.add_argument(
"--db",
default="amo_data.db",
help="Database file path (default: amo_data.db)"
)
backup_parser.add_argument(
"--keep",
type=int,
default=7,
help="Number of backups to keep (default: 7)"
)
# List command
list_parser = subparsers.add_parser("list", help="List available backups")
list_parser.add_argument(
"--db",
default="amo_data",
help="Database name without extension (default: amo_data)"
)
# Restore command
restore_parser = subparsers.add_parser("restore", help="Restore from a backup")
restore_parser.add_argument(
"backup_name",
help="Name of the backup file to restore"
)
restore_parser.add_argument(
"--db",
default="amo_data.db",
help="Database file path (default: amo_data.db)"
)
args = parser.parse_args()
if args.command == "backup":
backup_database(args.db, args.keep)
elif args.command == "list":
list_backups(args.db)
elif args.command == "restore":
restore_backup(args.backup_name, args.db)
else:
parser.print_help()
if __name__ == "__main__":
main()