fix(types): T4 — stop spreading TopologySummary as dict; fix heartbeat .get() and scope param

This commit is contained in:
2026-05-01 01:51:43 -04:00
parent 0f90dcfd3e
commit d187304e99
6 changed files with 11 additions and 8 deletions

View File

@@ -16,6 +16,7 @@ from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from collections.abc import MutableMapping
from typing import Any, Optional
from fastapi import APIRouter, Depends, HTTPException, Request
@@ -38,7 +39,7 @@ class HeartbeatRequest(BaseModel):
topology: Optional[dict[str, Any]] = None
def _extract_peer_fingerprint(scope: dict[str, Any]) -> Optional[str]:
def _extract_peer_fingerprint(scope: MutableMapping[str, Any]) -> Optional[str]:
"""Pull the peer cert's SHA-256 fingerprint from an ASGI scope.
Tries two extraction paths because uvicorn has historically stashed
@@ -123,7 +124,7 @@ async def _reconcile_topology_report(
except Exception:
log.exception("heartbeat: could not list active topologies")
return
mine = [t for t in topos if t.get("target_host_uuid") == host_uuid]
mine = [t for t in topos if t.target_host_uuid == host_uuid]
if not mine:
return

View File

@@ -120,4 +120,4 @@ async def api_create_blank_topology(
row = await repo.get_topology(topology_id)
if row is None: # pragma: no cover — create then vanish
raise HTTPException(status_code=500, detail="topology insert vanished")
return TopologySummary(**row)
return row

View File

@@ -74,4 +74,6 @@ async def api_create_topology(
) from exc
raise
row = await repo.get_topology(topology_id)
return TopologySummary(**row)
if row is None: # pragma: no cover — create then vanish
raise HTTPException(status_code=500, detail="topology insert vanished")
return row

View File

@@ -73,4 +73,4 @@ async def api_deploy_topology(
)
background.add_task(_run_deploy, topology_id)
return TopologySummary(**topo)
return topo

View File

@@ -6,7 +6,7 @@ from typing import Optional
from fastapi import APIRouter, Depends, Query
from decnet.telemetry import traced as _traced
from decnet.web.db.models import TopologyListResponse, TopologySummary
from decnet.web.db.models import TopologyListResponse
from decnet.web.dependencies import repo, require_viewer
router = APIRouter()
@@ -35,5 +35,5 @@ async def api_list_topologies(
total=total,
limit=limit,
offset=offset,
data=[TopologySummary(**r) for r in rows],
data=rows,
)

View File

@@ -76,4 +76,4 @@ async def api_teardown_topology(
)
background.add_task(_run_teardown, topology_id)
return TopologySummary(**topo)
return topo