Template
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""PostgreSQL backup/restore for the panel database."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
from .connection import get_pg_connection_params
|
|
from .store import invalidate_data_cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _pg_cli_env(password: str) -> dict[str, str]:
|
|
env = os.environ.copy()
|
|
if password:
|
|
env['PGPASSWORD'] = password
|
|
return env
|
|
|
|
|
|
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."""
|
|
params = get_pg_connection_params()
|
|
proc = subprocess.run(
|
|
[
|
|
'pg_dump',
|
|
'-h', params['host'],
|
|
'-p', params['port'],
|
|
'-U', params['user'],
|
|
'-d', params['dbname'],
|
|
'--no-owner',
|
|
'--no-acl',
|
|
'--clean',
|
|
'--if-exists',
|
|
],
|
|
capture_output=True,
|
|
check=False,
|
|
env=_pg_cli_env(params['password']),
|
|
)
|
|
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')
|
|
params = get_pg_connection_params()
|
|
proc = subprocess.run(
|
|
[
|
|
'psql',
|
|
'-h', params['host'],
|
|
'-p', params['port'],
|
|
'-U', params['user'],
|
|
'-d', params['dbname'],
|
|
'-v', 'ON_ERROR_STOP=1',
|
|
'-q',
|
|
],
|
|
input=data,
|
|
capture_output=True,
|
|
check=False,
|
|
env=_pg_cli_env(params['password']),
|
|
)
|
|
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')
|