130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
"""FC-3h: backup/restore Celery tasks.
|
|
|
|
All tasks live on the maintenance queue (per celery_app.task_routes).
|
|
task_run lifecycle tracking is automatic via FC-3i signals — these
|
|
tasks just record the operator-facing artifact metadata into
|
|
BackupRun.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from celery.exceptions import SoftTimeLimitExceeded
|
|
from sqlalchemy.exc import DBAPIError, OperationalError
|
|
|
|
from ..celery_app import celery
|
|
from ..config import get_config
|
|
from ..models import BackupRun
|
|
from ..services import backup_service
|
|
from ._sync_engine import sync_session_factory as _sync_session_factory
|
|
|
|
log = logging.getLogger(__name__)
|
|
IMAGES_ROOT = Path("/images")
|
|
|
|
|
|
def _mark_failed(session, row: BackupRun, exc: BaseException) -> None:
|
|
"""Flip a BackupRun row from running/restoring to error with a
|
|
truncated error message and finished_at. Caller already holds the
|
|
session open."""
|
|
row.status = "error"
|
|
row.error = f"{type(exc).__name__}: {exc}"[:2000]
|
|
row.finished_at = datetime.now(UTC)
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.backup.backup_db_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=10, retry_backoff_max=120, max_retries=2,
|
|
soft_time_limit=600, time_limit=720,
|
|
)
|
|
def backup_db_task(self, *, tag: str | None = None,
|
|
triggered_by: str = "manual") -> dict:
|
|
"""Create one DB backup. Returns {'backup_run_id': N}."""
|
|
SessionLocal = _sync_session_factory()
|
|
cfg = get_config()
|
|
now = datetime.now(UTC)
|
|
with SessionLocal() as session:
|
|
row = BackupRun(
|
|
kind="db", status="running", tag=tag,
|
|
triggered_by=triggered_by, started_at=now, manifest={},
|
|
)
|
|
session.add(row); session.commit(); session.refresh(row)
|
|
run_id = row.id
|
|
|
|
try:
|
|
result = backup_service.backup_db(
|
|
db_url=cfg.database_url_sync, images_root=IMAGES_ROOT,
|
|
tag=tag, triggered_by=triggered_by,
|
|
)
|
|
except (SoftTimeLimitExceeded, Exception) as exc:
|
|
with SessionLocal() as session:
|
|
row = session.get(BackupRun, run_id)
|
|
if row is not None:
|
|
_mark_failed(session, row, exc)
|
|
raise
|
|
|
|
with SessionLocal() as session:
|
|
row = session.get(BackupRun, run_id)
|
|
row.status = "ok"
|
|
row.finished_at = datetime.now(UTC)
|
|
row.sql_path = result["sql_path"]
|
|
row.size_bytes = result["size_bytes"]
|
|
row.manifest = {
|
|
"manifest_path": result["manifest_path"],
|
|
"ts": result["ts"],
|
|
}
|
|
session.commit()
|
|
return {"backup_run_id": run_id}
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.backup.backup_images_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=30, retry_backoff_max=300, max_retries=1,
|
|
soft_time_limit=21600, time_limit=23400,
|
|
)
|
|
def backup_images_task(self, *, tag: str | None = None,
|
|
triggered_by: str = "manual") -> dict:
|
|
"""Create one images backup. Same shape as backup_db_task; uses
|
|
tar_path instead of sql_path."""
|
|
SessionLocal = _sync_session_factory()
|
|
now = datetime.now(UTC)
|
|
with SessionLocal() as session:
|
|
row = BackupRun(
|
|
kind="images", status="running", tag=tag,
|
|
triggered_by=triggered_by, started_at=now, manifest={},
|
|
)
|
|
session.add(row); session.commit(); session.refresh(row)
|
|
run_id = row.id
|
|
|
|
try:
|
|
result = backup_service.backup_images(
|
|
images_root=IMAGES_ROOT,
|
|
tag=tag, triggered_by=triggered_by,
|
|
)
|
|
except (SoftTimeLimitExceeded, Exception) as exc:
|
|
with SessionLocal() as session:
|
|
row = session.get(BackupRun, run_id)
|
|
if row is not None:
|
|
_mark_failed(session, row, exc)
|
|
raise
|
|
|
|
with SessionLocal() as session:
|
|
row = session.get(BackupRun, run_id)
|
|
row.status = "ok"
|
|
row.finished_at = datetime.now(UTC)
|
|
row.tar_path = result["tar_path"]
|
|
row.size_bytes = result["size_bytes"]
|
|
row.manifest = {
|
|
"manifest_path": result["manifest_path"],
|
|
"ts": result["ts"],
|
|
}
|
|
session.commit()
|
|
return {"backup_run_id": run_id}
|