refactor(swarm): move router DTOs into decnet/web/db/models.py

_schemas.py was a local exception to the codebase convention. The rest
of the app keeps all API request/response DTOs in decnet/web/db/models.py
alongside UserResponse, DeployIniRequest, etc. — the swarm endpoints now
follow the same convention (SwarmEnrollRequest, SwarmHostView, etc).
Deletes decnet/web/router/swarm/_schemas.py.
This commit is contained in:
2026-04-18 19:28:15 -04:00
parent 811136e600
commit e2d6f857b5
8 changed files with 114 additions and 119 deletions

View File

@@ -4,7 +4,7 @@ from sqlalchemy import Column, Text
from sqlalchemy.dialects.mysql import MEDIUMTEXT
from sqlmodel import SQLModel, Field
from pydantic import BaseModel, ConfigDict, Field as PydanticField, BeforeValidator
from decnet.models import IniContent
from decnet.models import IniContent, DecnetConfig
# Use on columns that accumulate over an attacker's lifetime (commands,
# fingerprints, state blobs). TEXT on MySQL caps at 64 KiB; MEDIUMTEXT
@@ -265,3 +265,80 @@ class ComponentHealth(BaseModel):
class HealthResponse(BaseModel):
status: Literal["healthy", "degraded", "unhealthy"]
components: dict[str, ComponentHealth]
# --- Swarm API DTOs ---
# Request/response contracts for the master-side swarm controller
# (decnet/web/swarm_api.py). The underlying SQLModel tables — SwarmHost and
# DeckyShard — live above; these are the HTTP-facing shapes.
class SwarmEnrollRequest(BaseModel):
name: str = PydanticField(..., min_length=1, max_length=128)
address: str = PydanticField(..., description="IP or DNS the master uses to reach the worker")
agent_port: int = PydanticField(default=8765, ge=1, le=65535)
sans: list[str] = PydanticField(
default_factory=list,
description="Extra SANs (IPs / hostnames) to embed in the worker cert",
)
notes: Optional[str] = None
class SwarmEnrolledBundle(BaseModel):
"""Cert bundle returned to the operator — must be delivered to the worker."""
host_uuid: str
name: str
address: str
agent_port: int
fingerprint: str
ca_cert_pem: str
worker_cert_pem: str
worker_key_pem: str
class SwarmHostView(BaseModel):
uuid: str
name: str
address: str
agent_port: int
status: str
last_heartbeat: Optional[datetime] = None
client_cert_fingerprint: str
enrolled_at: datetime
notes: Optional[str] = None
class SwarmDeployRequest(BaseModel):
config: DecnetConfig
dry_run: bool = False
no_cache: bool = False
class SwarmTeardownRequest(BaseModel):
host_uuid: Optional[str] = PydanticField(
default=None,
description="If set, tear down only this worker; otherwise tear down all hosts",
)
decky_id: Optional[str] = None
class SwarmHostResult(BaseModel):
host_uuid: str
host_name: str
ok: bool
detail: Any | None = None
class SwarmDeployResponse(BaseModel):
results: list[SwarmHostResult]
class SwarmHostHealth(BaseModel):
host_uuid: str
name: str
address: str
reachable: bool
detail: Any | None = None
class SwarmCheckResponse(BaseModel):
results: list[SwarmHostHealth]

View File

@@ -1,82 +0,0 @@
"""Request/response models shared across the swarm router endpoints."""
from __future__ import annotations
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
from decnet.config import DecnetConfig
class EnrollRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=128)
address: str = Field(..., description="IP or DNS the master uses to reach the worker")
agent_port: int = Field(default=8765, ge=1, le=65535)
sans: list[str] = Field(
default_factory=list,
description="Extra SANs (IPs / hostnames) to embed in the worker cert",
)
notes: Optional[str] = None
class EnrolledBundle(BaseModel):
"""Cert bundle returned to the operator — must be delivered to the worker."""
host_uuid: str
name: str
address: str
agent_port: int
fingerprint: str
ca_cert_pem: str
worker_cert_pem: str
worker_key_pem: str
class SwarmHostView(BaseModel):
uuid: str
name: str
address: str
agent_port: int
status: str
last_heartbeat: Optional[datetime] = None
client_cert_fingerprint: str
enrolled_at: datetime
notes: Optional[str] = None
class DeployRequest(BaseModel):
config: DecnetConfig
dry_run: bool = False
no_cache: bool = False
class TeardownRequest(BaseModel):
host_uuid: str | None = Field(
default=None,
description="If set, tear down only this worker; otherwise tear down all hosts",
)
decky_id: str | None = None
class HostResult(BaseModel):
host_uuid: str
host_name: str
ok: bool
detail: Any | None = None
class DeployResponse(BaseModel):
results: list[HostResult]
class HostHealth(BaseModel):
host_uuid: str
name: str
address: str
reachable: bool
detail: Any | None = None
class CheckResponse(BaseModel):
results: list[HostHealth]

View File

@@ -15,20 +15,20 @@ from decnet.logging import get_logger
from decnet.swarm.client import AgentClient
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import CheckResponse, HostHealth
from decnet.web.db.models import SwarmCheckResponse, SwarmHostHealth
log = get_logger("swarm.check")
router = APIRouter()
@router.post("/check", response_model=CheckResponse, tags=["Swarm Health"])
@router.post("/check", response_model=SwarmCheckResponse, tags=["Swarm Health"])
async def api_check_hosts(
repo: BaseRepository = Depends(get_repo),
) -> CheckResponse:
) -> SwarmCheckResponse:
hosts = await repo.list_swarm_hosts()
async def _probe(host: dict[str, Any]) -> HostHealth:
async def _probe(host: dict[str, Any]) -> SwarmHostHealth:
try:
async with AgentClient(host=host) as agent:
body = await agent.health()
@@ -39,7 +39,7 @@ async def api_check_hosts(
"last_heartbeat": datetime.now(timezone.utc),
},
)
return HostHealth(
return SwarmHostHealth(
host_uuid=host["uuid"],
name=host["name"],
address=host["address"],
@@ -49,7 +49,7 @@ async def api_check_hosts(
except Exception as exc:
log.warning("swarm.check unreachable host=%s err=%s", host["name"], exc)
await repo.update_swarm_host(host["uuid"], {"status": "unreachable"})
return HostHealth(
return SwarmHostHealth(
host_uuid=host["uuid"],
name=host["name"],
address=host["address"],
@@ -58,4 +58,4 @@ async def api_check_hosts(
)
results = await asyncio.gather(*(_probe(h) for h in hosts))
return CheckResponse(results=list(results))
return SwarmCheckResponse(results=list(results))

View File

@@ -20,10 +20,10 @@ from decnet.logging import get_logger
from decnet.swarm.client import AgentClient
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import (
DeployRequest,
DeployResponse,
HostResult,
from decnet.web.db.models import (
SwarmDeployRequest,
SwarmDeployResponse,
SwarmHostResult,
)
log = get_logger("swarm.deploy")
@@ -47,11 +47,11 @@ def _worker_config(base: DecnetConfig, shard: list[DeckyConfig]) -> DecnetConfig
return base.model_copy(update={"deckies": shard})
@router.post("/deploy", response_model=DeployResponse, tags=["Swarm Deployments"])
@router.post("/deploy", response_model=SwarmDeployResponse, tags=["Swarm Deployments"])
async def api_deploy_swarm(
req: DeployRequest,
req: SwarmDeployRequest,
repo: BaseRepository = Depends(get_repo),
) -> DeployResponse:
) -> SwarmDeployResponse:
if req.config.mode != "swarm":
raise HTTPException(status_code=400, detail="mode must be 'swarm'")
@@ -64,7 +64,7 @@ async def api_deploy_swarm(
raise HTTPException(status_code=404, detail=f"unknown host_uuid: {host_uuid}")
hosts[host_uuid] = row
async def _dispatch(host_uuid: str, shard: list[DeckyConfig]) -> HostResult:
async def _dispatch(host_uuid: str, shard: list[DeckyConfig]) -> SwarmHostResult:
host = hosts[host_uuid]
cfg = _worker_config(req.config, shard)
try:
@@ -82,7 +82,7 @@ async def api_deploy_swarm(
}
)
await repo.update_swarm_host(host_uuid, {"status": "active"})
return HostResult(host_uuid=host_uuid, host_name=host["name"], ok=True, detail=body)
return SwarmHostResult(host_uuid=host_uuid, host_name=host["name"], ok=True, detail=body)
except Exception as exc:
log.exception("swarm.deploy dispatch failed host=%s", host["name"])
for d in shard:
@@ -96,9 +96,9 @@ async def api_deploy_swarm(
"updated_at": datetime.now(timezone.utc),
}
)
return HostResult(host_uuid=host_uuid, host_name=host["name"], ok=False, detail=str(exc))
return SwarmHostResult(host_uuid=host_uuid, host_name=host["name"], ok=False, detail=str(exc))
results = await asyncio.gather(
*(_dispatch(uuid_, shard) for uuid_, shard in buckets.items())
)
return DeployResponse(results=list(results))
return SwarmDeployResponse(results=list(results))

View File

@@ -18,21 +18,21 @@ from fastapi import APIRouter, Depends, HTTPException, status
from decnet.swarm import pki
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import EnrolledBundle, EnrollRequest
from decnet.web.db.models import SwarmEnrolledBundle, SwarmEnrollRequest
router = APIRouter()
@router.post(
"/enroll",
response_model=EnrolledBundle,
response_model=SwarmEnrolledBundle,
status_code=status.HTTP_201_CREATED,
tags=["Swarm Hosts"],
)
async def api_enroll_host(
req: EnrollRequest,
req: SwarmEnrollRequest,
repo: BaseRepository = Depends(get_repo),
) -> EnrolledBundle:
) -> SwarmEnrolledBundle:
existing = await repo.get_swarm_host_by_name(req.name)
if existing is not None:
raise HTTPException(status_code=409, detail=f"Worker '{req.name}' is already enrolled")
@@ -60,7 +60,7 @@ async def api_enroll_host(
"notes": req.notes,
}
)
return EnrolledBundle(
return SwarmEnrolledBundle(
host_uuid=host_uuid,
name=req.name,
address=req.address,

View File

@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import SwarmHostView
from decnet.web.db.models import SwarmHostView
router = APIRouter()

View File

@@ -7,7 +7,7 @@ from fastapi import APIRouter, Depends
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import SwarmHostView
from decnet.web.db.models import SwarmHostView
router = APIRouter()

View File

@@ -10,10 +10,10 @@ from decnet.logging import get_logger
from decnet.swarm.client import AgentClient
from decnet.web.db.repository import BaseRepository
from decnet.web.dependencies import get_repo
from decnet.web.router.swarm._schemas import (
DeployResponse,
HostResult,
TeardownRequest,
from decnet.web.db.models import (
SwarmDeployResponse,
SwarmHostResult,
SwarmTeardownRequest,
)
log = get_logger("swarm.teardown")
@@ -21,11 +21,11 @@ log = get_logger("swarm.teardown")
router = APIRouter()
@router.post("/teardown", response_model=DeployResponse, tags=["Swarm Deployments"])
@router.post("/teardown", response_model=SwarmDeployResponse, tags=["Swarm Deployments"])
async def api_teardown_swarm(
req: TeardownRequest,
req: SwarmTeardownRequest,
repo: BaseRepository = Depends(get_repo),
) -> DeployResponse:
) -> SwarmDeployResponse:
if req.host_uuid is not None:
row = await repo.get_swarm_host_by_uuid(req.host_uuid)
if row is None:
@@ -34,18 +34,18 @@ async def api_teardown_swarm(
else:
targets = await repo.list_swarm_hosts()
async def _call(host: dict[str, Any]) -> HostResult:
async def _call(host: dict[str, Any]) -> SwarmHostResult:
try:
async with AgentClient(host=host) as agent:
body = await agent.teardown(req.decky_id)
if req.decky_id is None:
await repo.delete_decky_shards_for_host(host["uuid"])
return HostResult(host_uuid=host["uuid"], host_name=host["name"], ok=True, detail=body)
return SwarmHostResult(host_uuid=host["uuid"], host_name=host["name"], ok=True, detail=body)
except Exception as exc:
log.exception("swarm.teardown failed host=%s", host["name"])
return HostResult(
return SwarmHostResult(
host_uuid=host["uuid"], host_name=host["name"], ok=False, detail=str(exc)
)
results = await asyncio.gather(*(_call(h) for h in targets))
return DeployResponse(results=list(results))
return SwarmDeployResponse(results=list(results))