backend/database_backup.py
Rama master · réplica de lectura del repositorio original.
"""Dialect-aware database backup with rotation and verification.
Supports both SQLite and PostgreSQL backends:
- **SQLite:** Uses the Online Backup API (``source.backup(target)``) for safe,
consistent backups without blocking writes.
- **PostgreSQL:** Uses ``pg_dump`` / ``pg_restore`` subprocess calls with
custom format (``-Fc``) for compressed backups — implemented in the
``_PostgresBackupMixin`` sibling module.
Backup rotation:
- daily: keep N most recent (default 7)
- weekly: keep N most recent (default 4)
- monthly: keep N most recent (default 3)
"""
import contextlib
import logging
import os
import re
import sqlite3
import threading
import time
from datetime import UTC, datetime
from database_backup_postgres import ( # noqa: F401 — re-exported for back-compat
_get_database_url,
_is_postgresql,
_parse_pg_url,
_PostgresBackupMixin,
)
from error_handler import DatabaseBackupError, DatabaseRestoreError
logger = logging.getLogger(__name__)
# How long a restore waits for other connections' write locks (milliseconds).
_SQLITE_RESTORE_BUSY_TIMEOUT_MS = 30_000
# Overall deadline for one copy. The busy timeout above only spaces out the
# retries; sqlite3's backup() itself keeps retrying a busy destination.
_SQLITE_RESTORE_DEADLINE_S = 60
def _page_size(path: str) -> int | None:
"""Page size of a SQLite file, or None when it cannot be read."""
try:
conn = sqlite3.connect(path)
try:
return int(conn.execute("PRAGMA page_size").fetchone()[0])
finally:
conn.close()
except sqlite3.Error:
return None
def _staged_with_page_size(source_path: str, page_size: int, workdir: str) -> str:
"""A copy of ``source_path`` rewritten to ``page_size``.
SQLite refuses to back up into a WAL database whose page size differs — and
reports it as "attempt to write a readonly database". VACUUM INTO writes a
fresh file with the page size set on the connection.
"""
staged = os.path.join(workdir, f"staged-{os.getpid()}-{int(datetime.now(UTC).timestamp())}.db")
conn = sqlite3.connect(source_path, isolation_level=None)
try:
conn.execute(f"PRAGMA page_size={int(page_size)}")
conn.execute("VACUUM INTO ?", (staged,))
finally:
conn.close()
return staged
def _sqlite_copy(source_path: str, target_path: str) -> None:
"""Copy one SQLite database onto another with the Online Backup API.
The whole copy runs in one step under SQLite's own locking, so ``target``
either holds all of ``source`` or is left unchanged, and connections other
threads keep open on ``target`` stay valid.
Two guards on top of the plain API call:
* ``timeout`` only feeds SQLite's busy handler, and ``backup()`` retries a
busy destination for as long as it takes. A deadline enforced from the
progress callback bounds the whole copy instead.
* A destination in WAL mode rejects a source with a different page size, so
the source is staged into a matching copy first.
"""
staged: str | None = None
if _page_size(source_path) != (target_page := _page_size(target_path)) and target_page:
staged = _staged_with_page_size(
source_path, target_page, os.path.dirname(target_path) or "."
)
source_path = staged
deadline = time.monotonic() + _SQLITE_RESTORE_DEADLINE_S
def _tick(status, remaining, total): # noqa: ARG001 — signature fixed by sqlite3
if time.monotonic() > deadline:
raise TimeoutError(
f"the database stayed busy for {_SQLITE_RESTORE_DEADLINE_S} s; nothing was changed"
)
try:
source = sqlite3.connect(source_path)
try:
target = sqlite3.connect(target_path, timeout=_SQLITE_RESTORE_BUSY_TIMEOUT_MS / 1000)
try:
source.backup(target, pages=512, progress=_tick)
finally:
target.close()
finally:
source.close()
finally:
if staged:
with contextlib.suppress(OSError):
os.remove(staged)
class DatabaseBackup(_PostgresBackupMixin):
"""Manages backup creation, verification, rotation, and restore.
Automatically dispatches to SQLite or PostgreSQL backend based on the
active SQLAlchemy dialect.
"""
def __init__(
self,
db_path: str,
backup_dir: str = "/config/backups",
retention_daily: int = 7,
retention_weekly: int = 4,
retention_monthly: int = 3,
) -> None:
self.db_path = db_path
self.backup_dir = backup_dir
self.retention_daily = retention_daily
self.retention_weekly = retention_weekly
self.retention_monthly = retention_monthly
os.makedirs(self.backup_dir, exist_ok=True)
# ── Dialect detection ─────────────────────────────────────────────────
@property
def is_postgresql(self) -> bool:
"""True if the active database backend is PostgreSQL."""
return _is_postgresql()
# ── Create backup ────────────────────────────────────────────────────
def create_backup(self, label: str = "daily") -> dict:
"""Create a verified backup.
Dispatches to SQLite backup API or pg_dump based on the active dialect.
Args:
label: Rotation bucket -- ``daily``, ``weekly``, or ``monthly``.
Returns:
Dict with ``path``, ``size_bytes``, ``verified``, ``timestamp``,
``label``, ``backend``.
Raises:
DatabaseBackupError: On any failure.
"""
if self.is_postgresql:
return self._backup_postgresql(label)
return self._backup_sqlite(label)
# ── SQLite backup ─────────────────────────────────────────────────────
def _backup_sqlite(self, label: str = "daily") -> dict:
"""Create a backup using the SQLite backup API."""
timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
filename = f"sublarr_{label}_{timestamp}.db"
dest = os.path.join(self.backup_dir, filename)
try:
source = sqlite3.connect(self.db_path)
target = sqlite3.connect(dest)
with target:
source.backup(target)
source.close()
target.close()
except Exception as exc:
# Clean up partial file
if os.path.exists(dest):
with contextlib.suppress(OSError):
os.remove(dest)
raise DatabaseBackupError(
f"SQLite backup failed: {exc}",
context={"dest": dest},
) from exc
# Verify
verified = self._verify_sqlite_backup(dest)
size = os.path.getsize(dest)
logger.info("SQLite backup created: %s (%d bytes, verified=%s)", dest, size, verified)
return {
"path": dest,
"filename": filename,
"size_bytes": size,
"verified": verified,
"timestamp": timestamp,
"label": label,
"backend": "sqlite",
}
# ── Verify ───────────────────────────────────────────────────────────
@staticmethod
def _verify_sqlite_backup(path: str) -> bool:
"""Run ``PRAGMA integrity_check`` on a SQLite backup file."""
try:
conn = sqlite3.connect(path)
result = conn.execute("PRAGMA integrity_check").fetchone()
conn.close()
return result is not None and result[0] == "ok"
except Exception as exc:
logger.warning("Backup verification failed for %s: %s", path, exc)
return False
# Backward-compatible alias
_verify_backup = _verify_sqlite_backup
# ── List backups ─────────────────────────────────────────────────────
def list_backups(self) -> list[dict]:
"""Return metadata for every backup file in the backup directory.
Matches both SQLite (.db) and PostgreSQL (.pgdump) backup files.
"""
backups: list[dict] = []
pattern = re.compile(r"^sublarr_(daily|weekly|monthly|manual)_(\d{8}_\d{6})\.(db|pgdump)$")
if not os.path.isdir(self.backup_dir):
return backups
for name in sorted(os.listdir(self.backup_dir), reverse=True):
match = pattern.match(name)
if not match:
continue
path = os.path.join(self.backup_dir, name)
try:
size = os.path.getsize(path)
except OSError:
size = 0
ext = match.group(3)
backups.append(
{
"filename": name,
"path": path,
"label": match.group(1),
"timestamp": match.group(2),
"size_bytes": size,
"backend": "postgresql" if ext == "pgdump" else "sqlite",
}
)
return backups
# ── Restore ──────────────────────────────────────────────────────────
def restore_backup(self, backup_path: str) -> dict:
"""Restore a backup, dispatching based on file extension.
For SQLite (.db): Copies it into the live database with the backup API
after an integrity check.
For PostgreSQL (.pgdump): Uses pg_restore subprocess.
Creates a safety backup before restoring (SQLite only).
Args:
backup_path: Absolute path to the backup file.
Returns:
Dict with ``restored_from`` and ``backend``.
Raises:
DatabaseRestoreError: On any failure.
"""
if not os.path.exists(backup_path):
raise DatabaseRestoreError(
f"Backup file not found: {backup_path}",
context={"backup_path": backup_path},
)
if backup_path.endswith(".pgdump"):
return self._restore_postgresql(backup_path)
return self._restore_sqlite(backup_path)
def _restore_sqlite(self, backup_path: str) -> dict:
"""Restore a SQLite backup by replacing the current database file."""
# Verify the backup before restoring
if not self._verify_sqlite_backup(backup_path):
raise DatabaseRestoreError(
"Backup file failed integrity check",
context={"backup_path": backup_path},
troubleshooting="The backup file may be corrupted. Choose a different backup.",
)
# Safety backup of the current DB, through the backup API as well: a
# file copy of a WAL database misses whatever is still in the -wal file.
safety_path = self.db_path + ".pre_restore"
try:
_sqlite_copy(self.db_path, safety_path)
except Exception as exc:
raise DatabaseRestoreError(
f"Could not create safety backup: {exc}",
) from exc
# Restore into the live database through SQLite's Online Backup API
# instead of copying the file over it. The copy (and deleting -wal/-shm)
# happened under connections other threads still held: VM test of
# 1.14.4-rc.2 got HTTP 500, a thread still reading the old data and
# "disk I/O error" on the next fresh connection. The backup API takes
# SQLite's own locks, is all-or-nothing, and every connection sees the
# restored data once its current transaction ends. This process's
# session is released first so it holds no transaction open.
from services.database_restore import release_db_connections
release_db_connections()
try:
_sqlite_copy(backup_path, self.db_path)
except Exception as exc:
raise DatabaseRestoreError(
f"Restore failed, the database was not changed: {exc}",
context={"backup_path": backup_path},
troubleshooting="Another connection may be writing; try again in a moment.",
) from exc
logger.info(
"SQLite database restored from %s (safety backup at %s)", backup_path, safety_path
)
return {
"restored_from": backup_path,
"safety_backup": safety_path,
"backend": "sqlite",
}
# ── Rotation ─────────────────────────────────────────────────────────
def rotate(self) -> int:
"""Delete old backups exceeding retention limits.
Returns:
Number of files deleted.
"""
backups = self.list_backups()
limits = {
"daily": self.retention_daily,
"weekly": self.retention_weekly,
"monthly": self.retention_monthly,
}
deleted = 0
for label, limit in limits.items():
label_backups = [b for b in backups if b["label"] == label]
# Already sorted newest-first by list_backups
for old in label_backups[limit:]:
try:
os.remove(old["path"])
deleted += 1
logger.debug("Rotated out old backup: %s", old["filename"])
except OSError as exc:
logger.warning("Could not delete old backup %s: %s", old["filename"], exc)
if deleted:
logger.info("Backup rotation: deleted %d old backups", deleted)
return deleted
# ── Scheduled backup (lightweight timer loop) ────────────────────────────────
_scheduler_thread: threading.Thread | None = None
_scheduler_stop = threading.Event()
# Polling cadence used while auto-backup is disabled. Short enough that
# toggling `backup_auto_enabled` ON in the UI takes effect promptly without
# a restart, long enough to avoid wasted wakeups.
_DISABLED_POLL_SECS = 60
def _label_for(now: datetime) -> str:
"""Pick the rotation bucket for *now* (1st of month → monthly; Monday → weekly)."""
if now.day == 1:
return "monthly"
if now.weekday() == 0:
return "weekly"
return "daily"
def _read_auto_backup_settings() -> tuple[bool, int, bool, bool]:
"""Read the four `backup_auto_*` config fields with safe defaults.
Returns (enabled, interval_hours, on_startup, notify_on_failure).
Default fallbacks match config_settings.Settings.
"""
try:
from config import get_settings
s = get_settings()
enabled = bool(getattr(s, "backup_auto_enabled", False))
interval = max(1, int(getattr(s, "backup_auto_interval_hours", 24) or 24))
on_startup = bool(getattr(s, "backup_auto_on_startup", False))
notify = bool(getattr(s, "backup_notify_on_failure", True))
return enabled, interval, on_startup, notify
except Exception:
# Settings unavailable (e.g. import-time race): pretend disabled.
logger.debug("backup scheduler: settings unavailable", exc_info=True)
return False, 24, False, True
def _emit_backup_failure_notification(exc: Exception) -> None:
"""Best-effort send of an `error` event to Apprise. Never raises."""
try:
from notifier import send_notification
send_notification(
title="Sublarr backup failed",
body=f"Scheduled database backup failed: {exc}",
event_type="error",
)
except Exception:
logger.debug("backup scheduler: notify-on-failure emit failed", exc_info=True)
def _run_one_backup(backup: "DatabaseBackup", notify_on_failure: bool) -> None:
"""Run a single backup + rotation. Captures failures locally."""
try:
backup.create_backup(label=_label_for(datetime.now(UTC)))
backup.rotate()
except Exception as exc:
logger.error("Scheduled backup failed: %s", exc, exc_info=True)
if notify_on_failure:
_emit_backup_failure_notification(exc)
def start_backup_scheduler(
db_path: str,
backup_dir: str = "/config/backups",
app=None,
) -> None:
"""Start a daemon thread that runs auto-backups when configured.
Honours four config fields (writable from the UI):
- `backup_auto_enabled` — if False, the loop polls and skips runs
- `backup_auto_interval_hours` — sleep duration between runs (>=1)
- `backup_auto_on_startup` — run one backup immediately on start
- `backup_notify_on_failure` — emit an `error` notification when
create_backup raises
`app` is the Flask app; needed because `notifier.send_notification`
writes to `notification_history` and other DB-touching helpers
require an app context.
"""
global _scheduler_thread
if _scheduler_thread and _scheduler_thread.is_alive():
return
_scheduler_stop.clear()
def _loop() -> None:
backup = DatabaseBackup(db_path, backup_dir)
ran_startup = False
while not _scheduler_stop.is_set():
enabled, interval_hours, on_startup, notify_on_failure = _read_auto_backup_settings()
if not enabled:
# Auto-backup turned off — poll the toggle without burning
# CPU. Reset on_startup tracking so re-enabling re-fires.
ran_startup = False
if _scheduler_stop.wait(timeout=_DISABLED_POLL_SECS):
break
continue
if on_startup and not ran_startup:
ran_startup = True
if app is not None:
with app.app_context():
_run_one_backup(backup, notify_on_failure)
else:
_run_one_backup(backup, notify_on_failure)
# Fall through into the regular interval wait.
wait_secs = max(1, interval_hours) * 3600
if _scheduler_stop.wait(timeout=wait_secs):
break
# Re-read the gate *after* the long sleep so a user who turned
# auto-backup OFF mid-sleep doesn't get one final stale run.
enabled, _interval_hours, _on_startup, notify_on_failure = _read_auto_backup_settings()
if not enabled:
continue
if app is not None:
with app.app_context():
_run_one_backup(backup, notify_on_failure)
else:
_run_one_backup(backup, notify_on_failure)
_scheduler_thread = threading.Thread(target=_loop, daemon=True, name="backup-scheduler")
_scheduler_thread.start()
logger.info("Backup scheduler started (config-driven)")
def stop_backup_scheduler() -> None:
"""Stop the backup scheduler thread."""
_scheduler_stop.set()