Template
60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
"""PostgreSQL backup/restore for the panel database."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
from .connection import get_database_url
|
|
from .store import invalidate_data_cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def backup_filename() -> str:
|
|
stamp = datetime.now(timezone.utc).strftime('%Y-%m-%d_%H%M%S')
|
|
return f'amnezia_panel_backup_{stamp}.sql'
|
|
|
|
|
|
def export_database_sql() -> bytes:
|
|
"""Create a plain SQL dump of the panel PostgreSQL database."""
|
|
url = get_database_url()
|
|
proc = subprocess.run(
|
|
[
|
|
'pg_dump',
|
|
'--dbname', url,
|
|
'--no-owner',
|
|
'--no-acl',
|
|
'--clean',
|
|
'--if-exists',
|
|
],
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if proc.returncode != 0:
|
|
err = proc.stderr.decode('utf-8', errors='replace').strip()
|
|
raise RuntimeError(err or 'pg_dump failed')
|
|
if not proc.stdout:
|
|
raise RuntimeError('pg_dump returned empty dump')
|
|
return proc.stdout
|
|
|
|
|
|
def restore_database_sql(data: bytes) -> None:
|
|
"""Restore panel data from a plain SQL dump produced by pg_dump."""
|
|
if not data or not data.strip():
|
|
raise ValueError('Empty backup file')
|
|
url = get_database_url()
|
|
proc = subprocess.run(
|
|
['psql', '--dbname', url, '-v', 'ON_ERROR_STOP=1', '-q'],
|
|
input=data,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if proc.returncode != 0:
|
|
err = proc.stderr.decode('utf-8', errors='replace').strip()
|
|
out = proc.stdout.decode('utf-8', errors='replace').strip()
|
|
raise RuntimeError(err or out or 'psql restore failed')
|
|
invalidate_data_cache()
|
|
logger.info('PostgreSQL backup restored successfully')
|