90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
# scripts/backup_database.py
|
|
"""Database backup utility that uses project configuration."""
|
|
|
|
import os
|
|
import sys
|
|
import shutil
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
# Add project root to Python path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def get_database_path():
|
|
"""Extract database path from DATABASE_URL."""
|
|
db_url = settings.database_url
|
|
|
|
if db_url.startswith('sqlite:///'):
|
|
# Remove sqlite:/// prefix and handle relative paths
|
|
db_path = db_url.replace('sqlite:///', '')
|
|
if db_path.startswith('./'):
|
|
db_path = db_path[2:] # Remove ./ prefix
|
|
return db_path
|
|
else:
|
|
# For PostgreSQL or other databases, we can't do file backup
|
|
print(f"[INFO] Database type: {db_url.split('://')[0]}")
|
|
print("[ERROR] File backup only supported for SQLite databases")
|
|
return None
|
|
|
|
|
|
def backup_sqlite_database():
|
|
"""Backup SQLite database using proper connection."""
|
|
|
|
db_path = get_database_path()
|
|
if not db_path:
|
|
return False
|
|
|
|
if not os.path.exists(db_path):
|
|
print(f"[INFO] No database file found at: {db_path}")
|
|
print("[INFO] Nothing to backup")
|
|
return True
|
|
|
|
# Create backup 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"database_backup_{timestamp}.db"
|
|
|
|
try:
|
|
# Use SQLite backup API for better consistency
|
|
source_conn = sqlite3.connect(db_path)
|
|
backup_conn = sqlite3.connect(str(backup_path))
|
|
|
|
source_conn.backup(backup_conn)
|
|
|
|
source_conn.close()
|
|
backup_conn.close()
|
|
|
|
print(f"[SUCCESS] Database backed up to: {backup_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"[ERROR] Backup failed: {e}")
|
|
return False
|
|
|
|
|
|
def backup_database():
|
|
"""Main backup function that handles different database types."""
|
|
|
|
print("[BACKUP] Starting database backup...")
|
|
print(f"[INFO] Database URL: {settings.database_url}")
|
|
|
|
if settings.database_url.startswith('sqlite'):
|
|
return backup_sqlite_database()
|
|
else:
|
|
print("[INFO] For PostgreSQL databases, use pg_dump:")
|
|
print(f"pg_dump {settings.database_url} > backup_$(date +%Y%m%d_%H%M%S).sql")
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = backup_database()
|
|
sys.exit(0 if success else 1)
|