feat: add backup engine with full/incremental support
- BackupEngine class executes jobs and creates backups - Automatic fallback from incremental to full if no parent exists - SHA-256 checksum calculation for integrity - Tests for full backup and incremental fallback
This commit is contained in:
@@ -0,0 +1,130 @@
|
|||||||
|
import os
|
||||||
|
import hashlib
|
||||||
|
import shutil
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
from sqlalchemy import select
|
||||||
|
from app.models import Job, JobExecution, Backup
|
||||||
|
from backup.adapters import get_adapter
|
||||||
|
|
||||||
|
class BackupEngine:
|
||||||
|
def __init__(self, db: AsyncSession):
|
||||||
|
self.db = db
|
||||||
|
|
||||||
|
async def execute_job(self, job_id: int, triggered_by: str = "manual") -> JobExecution:
|
||||||
|
# Create execution record
|
||||||
|
execution = JobExecution(
|
||||||
|
job_id=job_id,
|
||||||
|
status="pending",
|
||||||
|
triggered_by=triggered_by
|
||||||
|
)
|
||||||
|
self.db.add(execution)
|
||||||
|
await self.db.commit()
|
||||||
|
await self.db.refresh(execution)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Load job with source
|
||||||
|
result = await self.db.execute(
|
||||||
|
select(Job).where(Job.id == job_id)
|
||||||
|
)
|
||||||
|
job = result.scalar_one()
|
||||||
|
|
||||||
|
# Update status to running
|
||||||
|
execution.status = "running"
|
||||||
|
execution.started_at = datetime.now(timezone.utc)
|
||||||
|
await self.db.commit()
|
||||||
|
|
||||||
|
# Determine strategy
|
||||||
|
strategy = job.strategy
|
||||||
|
parent_backup_id = None
|
||||||
|
|
||||||
|
if strategy == "incremental":
|
||||||
|
# Find last successful full backup
|
||||||
|
result = await self.db.execute(
|
||||||
|
select(Backup)
|
||||||
|
.join(JobExecution)
|
||||||
|
.where(
|
||||||
|
JobExecution.job_id == job_id,
|
||||||
|
JobExecution.status == "success",
|
||||||
|
Backup.type == "full"
|
||||||
|
)
|
||||||
|
.order_by(Backup.created_at.desc())
|
||||||
|
)
|
||||||
|
last_full = result.scalar_one_or_none()
|
||||||
|
|
||||||
|
if last_full:
|
||||||
|
parent_backup_id = last_full.id
|
||||||
|
else:
|
||||||
|
# No full backup exists, do full instead
|
||||||
|
strategy = "full"
|
||||||
|
|
||||||
|
# Create backup directory
|
||||||
|
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H%M%S")
|
||||||
|
backup_dir = Path(job.destination_path) / str(job_id) / f"{timestamp}_{strategy}"
|
||||||
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
# Get adapter and connect
|
||||||
|
adapter = get_adapter(job.source.type, job.source.config)
|
||||||
|
await adapter.connect()
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Copy files
|
||||||
|
total_processed = 0
|
||||||
|
total_backed_up = 0
|
||||||
|
|
||||||
|
source_path = Path(job.source.config.get("path", "."))
|
||||||
|
|
||||||
|
for item in source_path.rglob("*"):
|
||||||
|
if item.is_file():
|
||||||
|
rel_path = item.relative_to(source_path)
|
||||||
|
dest_path = backup_dir / "data" / rel_path
|
||||||
|
dest_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
# Copy file
|
||||||
|
shutil.copy2(item, dest_path)
|
||||||
|
|
||||||
|
size = item.stat().st_size
|
||||||
|
total_processed += size
|
||||||
|
total_backed_up += size
|
||||||
|
|
||||||
|
# Calculate checksum
|
||||||
|
checksum = await self._calculate_checksum(backup_dir)
|
||||||
|
|
||||||
|
# Create backup record
|
||||||
|
backup = Backup(
|
||||||
|
execution_id=execution.id,
|
||||||
|
storage_path=str(backup_dir),
|
||||||
|
size_bytes=total_backed_up,
|
||||||
|
checksum=checksum,
|
||||||
|
type=strategy,
|
||||||
|
parent_backup_id=parent_backup_id
|
||||||
|
)
|
||||||
|
self.db.add(backup)
|
||||||
|
|
||||||
|
# Update execution
|
||||||
|
execution.status = "success"
|
||||||
|
execution.completed_at = datetime.now(timezone.utc)
|
||||||
|
execution.bytes_processed = total_processed
|
||||||
|
execution.bytes_backed_up = total_backed_up
|
||||||
|
|
||||||
|
finally:
|
||||||
|
await adapter.disconnect()
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
execution.status = "failed"
|
||||||
|
execution.completed_at = datetime.now(timezone.utc)
|
||||||
|
execution.error_message = str(e)
|
||||||
|
|
||||||
|
await self.db.commit()
|
||||||
|
return execution
|
||||||
|
|
||||||
|
async def _calculate_checksum(self, path: Path) -> str:
|
||||||
|
hasher = hashlib.sha256()
|
||||||
|
for item in sorted(path.rglob("*")):
|
||||||
|
if item.is_file():
|
||||||
|
with open(item, "rb") as f:
|
||||||
|
while chunk := f.read(8192):
|
||||||
|
hasher.update(chunk)
|
||||||
|
return hasher.hexdigest()
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
import pytest
|
||||||
|
import tempfile
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
from app.models import Source, Job, JobExecution
|
||||||
|
from backup.engine import BackupEngine
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def test_source(db: AsyncSession):
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
# Create test files
|
||||||
|
(Path(tmpdir) / "test.txt").write_text("Hello, World!")
|
||||||
|
(Path(tmpdir) / "subdir").mkdir()
|
||||||
|
(Path(tmpdir) / "subdir" / "nested.txt").write_text("Nested content")
|
||||||
|
|
||||||
|
source = Source(
|
||||||
|
name="Test Source",
|
||||||
|
type="local",
|
||||||
|
config={"path": tmpdir}
|
||||||
|
)
|
||||||
|
db.add(source)
|
||||||
|
await db.commit()
|
||||||
|
await db.refresh(source)
|
||||||
|
yield source
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def test_job(db: AsyncSession, test_source):
|
||||||
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
|
job = Job(
|
||||||
|
name="Test Job",
|
||||||
|
source_id=test_source.id,
|
||||||
|
strategy="full",
|
||||||
|
destination_path=tmpdir
|
||||||
|
)
|
||||||
|
db.add(job)
|
||||||
|
await db.commit()
|
||||||
|
await db.refresh(job)
|
||||||
|
yield job
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_execute_full_backup(db: AsyncSession, test_job):
|
||||||
|
engine = BackupEngine(db)
|
||||||
|
execution = await engine.execute_job(test_job.id, triggered_by="manual")
|
||||||
|
|
||||||
|
assert execution.status == "success"
|
||||||
|
assert execution.bytes_processed > 0
|
||||||
|
assert execution.bytes_backed_up > 0
|
||||||
|
assert execution.triggered_by == "manual"
|
||||||
|
|
||||||
|
# Verify backup was created
|
||||||
|
assert len(test_job.executions) == 1
|
||||||
|
backup = test_job.executions[0].backups[0]
|
||||||
|
assert backup.type == "full"
|
||||||
|
assert backup.checksum is not None
|
||||||
|
assert os.path.exists(backup.storage_path)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_execute_incremental_without_full(db: AsyncSession, test_job):
|
||||||
|
# Set job to incremental but no full backup exists
|
||||||
|
test_job.strategy = "incremental"
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
engine = BackupEngine(db)
|
||||||
|
execution = await engine.execute_job(test_job.id)
|
||||||
|
|
||||||
|
# Should fall back to full backup
|
||||||
|
assert execution.status == "success"
|
||||||
|
backup = execution.backups[0]
|
||||||
|
assert backup.type == "full"
|
||||||
|
assert backup.parent_backup_id is None
|
||||||
Reference in New Issue
Block a user