135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.dependencies import get_current_user, get_db
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
from app.models.snapshot import ConfigSnapshot
|
|
from app.models.user import User
|
|
from app.schemas.snapshot import SnapshotBrief, SnapshotDetail, SnapshotDiffResponse
|
|
from app.services import snapshot as snapshot_service
|
|
|
|
router = APIRouter(prefix="/api/v1/objects/{obj_id}/snapshots", tags=["snapshots"])
|
|
|
|
|
|
async def _get_object(db: AsyncSession, obj_id: int) -> Object:
|
|
obj = await db.get(Object, obj_id)
|
|
if obj is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Object not found")
|
|
return obj
|
|
|
|
|
|
async def _get_snapshot_for_object(
|
|
db: AsyncSession, obj_id: int, snapshot_id: int
|
|
) -> ConfigSnapshot:
|
|
"""Load a snapshot, verifying it belongs to a device on obj_id."""
|
|
result = await db.execute(
|
|
select(ConfigSnapshot)
|
|
.join(Device, ConfigSnapshot.device_id == Device.id)
|
|
.options(
|
|
selectinload(ConfigSnapshot.snapshot_content),
|
|
selectinload(ConfigSnapshot.device),
|
|
)
|
|
.where(ConfigSnapshot.id == snapshot_id, Device.object_id == obj_id)
|
|
)
|
|
snap = result.scalar_one_or_none()
|
|
if snap is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Snapshot not found")
|
|
return snap
|
|
|
|
|
|
@router.get("", response_model=list[SnapshotBrief])
|
|
async def list_snapshots(
|
|
obj_id: int,
|
|
device_id: Optional[int] = Query(None),
|
|
label: Optional[str] = Query(None),
|
|
limit: int = Query(100, ge=1, le=500),
|
|
skip: int = Query(0, ge=0),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
await _get_object(db, obj_id)
|
|
|
|
query = (
|
|
select(ConfigSnapshot)
|
|
.join(Device, ConfigSnapshot.device_id == Device.id)
|
|
.options(
|
|
selectinload(ConfigSnapshot.snapshot_content),
|
|
selectinload(ConfigSnapshot.device),
|
|
)
|
|
.where(Device.object_id == obj_id)
|
|
.order_by(ConfigSnapshot.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
|
|
if device_id is not None:
|
|
query = query.where(ConfigSnapshot.device_id == device_id)
|
|
if label is not None:
|
|
query = query.where(ConfigSnapshot.label.ilike(f"%{label}%"))
|
|
|
|
result = await db.execute(query)
|
|
snaps = result.scalars().all()
|
|
|
|
out = []
|
|
for s in snaps:
|
|
out.append(
|
|
SnapshotBrief(
|
|
id=s.id,
|
|
device_id=s.device_id,
|
|
device_ip=s.device.ip if s.device else None,
|
|
device_hostname=s.device.hostname if s.device else None,
|
|
sha256=s.sha256,
|
|
label=s.label,
|
|
size=s.snapshot_content.size,
|
|
created_at=s.created_at,
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
@router.get("/diff", response_model=SnapshotDiffResponse)
|
|
async def diff_snapshots(
|
|
obj_id: int,
|
|
a: int = Query(..., description="First snapshot ID"),
|
|
b: int = Query(..., description="Second snapshot ID"),
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
await _get_object(db, obj_id)
|
|
# Verify both snapshots belong to this object
|
|
await _get_snapshot_for_object(db, obj_id, a)
|
|
await _get_snapshot_for_object(db, obj_id, b)
|
|
|
|
diff_text, is_identical = await snapshot_service.compute_diff(a, b)
|
|
return SnapshotDiffResponse(
|
|
snapshot_a_id=a,
|
|
snapshot_b_id=b,
|
|
diff=diff_text,
|
|
is_identical=is_identical,
|
|
)
|
|
|
|
|
|
@router.get("/{snapshot_id}", response_model=SnapshotDetail)
|
|
async def get_snapshot(
|
|
obj_id: int,
|
|
snapshot_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
snap = await _get_snapshot_for_object(db, obj_id, snapshot_id)
|
|
return SnapshotDetail(
|
|
id=snap.id,
|
|
device_id=snap.device_id,
|
|
device_ip=snap.device.ip if snap.device else None,
|
|
device_hostname=snap.device.hostname if snap.device else None,
|
|
sha256=snap.sha256,
|
|
label=snap.label,
|
|
size=snap.snapshot_content.size,
|
|
created_at=snap.created_at,
|
|
content=snap.snapshot_content.content,
|
|
)
|