106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
"""
|
|
SnapshotService — content-addressed storage for device config snapshots.
|
|
|
|
SHA-256 deduplication: identical content is stored once in snapshot_contents.
|
|
Each capture creates a new config_snapshots row pointing to the content.
|
|
"""
|
|
|
|
import difflib
|
|
import hashlib
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.snapshot import ConfigSnapshot, SnapshotContent
|
|
from app.services.database import AsyncSessionLocal
|
|
|
|
|
|
async def save_snapshot(
|
|
device_id: int,
|
|
content: str,
|
|
label: Optional[str] = None,
|
|
) -> tuple[int, str, int]:
|
|
"""
|
|
Persist a snapshot, deduplicating content by SHA-256.
|
|
|
|
Returns (snapshot_id, sha256, size).
|
|
"""
|
|
raw = content.encode("utf-8", errors="replace")
|
|
sha256 = hashlib.sha256(raw).hexdigest()
|
|
size = len(raw)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
async with db.begin():
|
|
# Insert content only if it doesn't exist yet (dedup)
|
|
existing = await db.get(SnapshotContent, sha256)
|
|
if existing is None:
|
|
db.add(SnapshotContent(sha256=sha256, content=content, size=size, created_at=now))
|
|
|
|
snap = ConfigSnapshot(
|
|
device_id=device_id,
|
|
sha256=sha256,
|
|
label=label,
|
|
created_at=now,
|
|
)
|
|
db.add(snap)
|
|
await db.flush()
|
|
snap_id = snap.id
|
|
|
|
return snap_id, sha256, size
|
|
|
|
|
|
async def get_snapshot_content(snapshot_id: int) -> Optional[str]:
|
|
"""Load content for a single snapshot."""
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(
|
|
select(ConfigSnapshot)
|
|
.options(selectinload(ConfigSnapshot.snapshot_content))
|
|
.where(ConfigSnapshot.id == snapshot_id)
|
|
)
|
|
snap = result.scalar_one_or_none()
|
|
if snap is None:
|
|
return None
|
|
return snap.snapshot_content.content
|
|
|
|
|
|
async def compute_diff(snapshot_id_a: int, snapshot_id_b: int) -> tuple[str, bool]:
|
|
"""
|
|
Compute unified diff between two snapshots.
|
|
|
|
Returns (diff_text, is_identical).
|
|
"""
|
|
async with AsyncSessionLocal() as db:
|
|
result_a = await db.execute(
|
|
select(ConfigSnapshot)
|
|
.options(selectinload(ConfigSnapshot.snapshot_content))
|
|
.where(ConfigSnapshot.id == snapshot_id_a)
|
|
)
|
|
snap_a = result_a.scalar_one_or_none()
|
|
|
|
result_b = await db.execute(
|
|
select(ConfigSnapshot)
|
|
.options(selectinload(ConfigSnapshot.snapshot_content))
|
|
.where(ConfigSnapshot.id == snapshot_id_b)
|
|
)
|
|
snap_b = result_b.scalar_one_or_none()
|
|
|
|
if snap_a is None or snap_b is None:
|
|
return "", False
|
|
|
|
content_a = snap_a.snapshot_content.content
|
|
content_b = snap_b.snapshot_content.content
|
|
|
|
if content_a == content_b:
|
|
return "", True
|
|
|
|
lines_a = content_a.splitlines(keepends=True)
|
|
lines_b = content_b.splitlines(keepends=True)
|
|
|
|
label_a = f"#{snap_a.id} {snap_a.label or snap_a.sha256[:12]}"
|
|
label_b = f"#{snap_b.id} {snap_b.label or snap_b.sha256[:12]}"
|
|
|
|
diff = difflib.unified_diff(lines_a, lines_b, fromfile=label_a, tofile=label_b)
|
|
return "".join(diff), False
|