feat: add response models for endpoints returning untyped objects

gns3server/api/routes/controller/acl.py

@@ -80,7 +80,11 @@ def _iter_route_paths(routes: Sequence[BaseRoute], prefix: str = "", include_mou
                 yield from _iter_route_paths(mounted_routes, _join_paths(prefix, route.path), include_mounted_routes)
 
 
-@router.get("/endpoints", dependencies=[Depends(has_privilege("ACE.Audit"))])
+@router.get(
+    "/endpoints",
+    response_model=List[schemas.ACEEndpoint],
+    dependencies=[Depends(has_privilege("ACE.Audit"))],
+)
 async def endpoints(
     users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),

gns3server/api/routes/controller/api_keys.py

@@ -50,12 +50,12 @@ def _generate_api_key(api_key_id: UUID | None = None) -> tuple[str, str, str, UU
     return raw_key, key_hash, key_prefix, api_key_id
 
 
-@router.post("", status_code=status.HTTP_201_CREATED)
+@router.post("", response_model=schemas.ApiKeyCreated, status_code=status.HTTP_201_CREATED)
 async def create_api_key(
     api_key_data: schemas.ApiKeyCreate,
     current_user: schemas.User = Depends(get_current_active_user),
     api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
-) -> dict:
+) -> schemas.ApiKeyCreated:
     """Create a new API key. The full key is returned only once."""
 
     raw_key, key_hash, key_prefix, new_key_id = _generate_api_key()
@@ -66,42 +66,42 @@ async def create_api_key(
         key_hash=key_hash,
         key_prefix=key_prefix,
     )
-    return {
-        "api_key_id": str(db_key.api_key_id),
-        "api_key": raw_key,
-        "name": db_key.name,
-        "key_prefix": db_key.key_prefix,
-        "created_at": db_key.created_at.isoformat() if db_key.created_at else None,
-    }
+    return schemas.ApiKeyCreated(
+        api_key_id=db_key.api_key_id,
+        api_key=raw_key,
+        name=db_key.name,
+        key_prefix=db_key.key_prefix,
+        created_at=db_key.created_at.isoformat() if db_key.created_at else None,
+    )
 
 
-@router.get("")
+@router.get("", response_model=list[schemas.ApiKey])
 async def list_api_keys(
     current_user: schemas.User = Depends(get_current_active_user),
     api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
-) -> list[dict]:
+) -> list[schemas.ApiKey]:
     """List all API keys for the current user."""
 
     keys = await api_keys_repo.get_api_keys_by_user(current_user.user_id)
     return [
-        {
-            "api_key_id": str(k.api_key_id),
-            "name": k.name,
-            "key_prefix": k.key_prefix,
-            "created_at": k.created_at.isoformat() if k.created_at else None,
-            "last_used_at": k.last_used_at.isoformat() if k.last_used_at else None,
-            "revoked": k.revoked,
-        }
+        schemas.ApiKey(
+            api_key_id=k.api_key_id,
+            name=k.name,
+            key_prefix=k.key_prefix,
+            created_at=k.created_at.isoformat() if k.created_at else None,
+            last_used_at=k.last_used_at.isoformat() if k.last_used_at else None,
+            revoked=k.revoked,
+        )
         for k in keys
     ]
 
 
-@router.post("/{api_key_id}/revoke", status_code=status.HTTP_200_OK)
+@router.post("/{api_key_id}/revoke", response_model=schemas.ApiKeyMessage, status_code=status.HTTP_200_OK)
 async def revoke_api_key(
     api_key_id: UUID,
     current_user: schemas.User = Depends(get_current_active_user),
     api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
-) -> dict:
+) -> schemas.ApiKeyMessage:
     """Revoke an API key. It will immediately stop working, but can be restored."""
 
     key = await api_keys_repo.get_api_key(api_key_id)
@@ -111,15 +111,15 @@ async def revoke_api_key(
         raise HTTPException(status_code=403, detail="Cannot modify another user's API key")
 
     await api_keys_repo.revoke_api_key(api_key_id)
-    return {"message": f"API key '{key.name}' revoked"}
+    return schemas.ApiKeyMessage(message=f"API key '{key.name}' revoked")
 
 
-@router.post("/{api_key_id}/restore", status_code=status.HTTP_200_OK)
+@router.post("/{api_key_id}/restore", response_model=schemas.ApiKeyMessage, status_code=status.HTTP_200_OK)
 async def restore_api_key(
     api_key_id: UUID,
     current_user: schemas.User = Depends(get_current_active_user),
     api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
-) -> dict:
+) -> schemas.ApiKeyMessage:
     """Restore a previously revoked API key."""
 
     key = await api_keys_repo.get_api_key(api_key_id)
@@ -129,7 +129,7 @@ async def restore_api_key(
         raise HTTPException(status_code=403, detail="Cannot modify another user's API key")
 
     await api_keys_repo.restore_api_key(api_key_id)
-    return {"message": f"API key '{key.name}' restored"}
+    return schemas.ApiKeyMessage(message=f"API key '{key.name}' restored")
 
 
 @router.delete("/{api_key_id}", status_code=status.HTTP_204_NO_CONTENT)

gns3server/api/routes/controller/appliances.py

@@ -82,6 +82,8 @@ def get_appliance(appliance_id: UUID) -> schemas.Appliance:
 
 @router.post(
     "/{appliance_id}/version",
+    response_model=schemas.Appliance,
+    response_model_exclude_unset=True,
     status_code=status.HTTP_201_CREATED,
     dependencies=[Depends(has_privilege("Appliance.Allocate"))],
 )

gns3server/api/routes/controller/computes.py

@@ -212,7 +212,7 @@ async def vmware_vms(compute_id: Union[str, UUID]) -> List[schemas.ComputeVMware
     return result
 
 
-@router.post("/{compute_id}/dynamips/auto_idlepc")
+@router.post("/{compute_id}/dynamips/auto_idlepc", response_model=schemas.IdlePC)
 async def dynamips_autoidlepc(compute_id: Union[str, UUID], auto_idle_pc: schemas.AutoIdlePC):
     """
     Find a suitable Idle-PC value for a given IOS image. This may take a few minutes.

gns3server/api/routes/controller/controller.py

@@ -161,7 +161,12 @@ async def update_iou_license(iou_license: schemas.IOULicense) -> schemas.IOULice
     return current_iou_license
 
 
-@router.get("/statistics", dependencies=[Depends(get_current_active_user)])
+@router.get(
+    "/statistics",
+    response_model=schemas.ServerStatistics,
+    response_model_exclude_unset=True,
+    dependencies=[Depends(get_current_active_user)],
+)
 async def statistics() -> dict:
     """
     Return server statistics including compute resources, projects, and nodes.

gns3server/api/routes/controller/gns3vm.py

@@ -30,7 +30,7 @@ from gns3server.controller import Controller
 router = APIRouter()
 
 
-@router.get("/engines")
+@router.get("/engines", response_model=List[schemas.GNS3VMEngineInfo])
 async def get_engines() -> List[dict]:
     """
     Return the list of supported engines for the GNS3VM.
@@ -40,7 +40,7 @@ async def get_engines() -> List[dict]:
     return gns3_vm.engine_list()
 
 
-@router.get("/engines/{engine}/vms")
+@router.get("/engines/{engine}/vms", response_model=List[schemas.GNS3VMInfo])
 async def get_vms(engine: str) -> List[dict]:
     """
     Return all the available VMs for a specific virtualization engine.

gns3server/api/routes/controller/images.py

@@ -327,7 +327,12 @@ async def prune_images(
     )
 
 
-@router.post("/install", status_code=status.HTTP_200_OK, dependencies=[Depends(has_privilege("Image.Allocate"))])
+@router.post(
+    "/install",
+    response_model=schemas.ImageInstallResult,
+    status_code=status.HTTP_200_OK,
+    dependencies=[Depends(has_privilege("Image.Allocate"))],
+)
 async def install_images(
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),

gns3server/api/routes/controller/links.py

@@ -20,7 +20,7 @@ API routes for links.
 
 import logging
 import os
-from typing import Any, List, Union
+from typing import Any, Dict, List, Union
 from uuid import UUID, uuid4
 
 import aiohttp
@@ -123,7 +123,12 @@ async def create_link(project_id: UUID, link_create: schemas.LinkCreate) -> sche
     return link.asdict()
 
 
-@router.get("/{link_id}/available_filters", dependencies=[Depends(has_privilege("Link.Audit"))])
+@router.get(
+    "/{link_id}/available_filters",
+    response_model=List[schemas.LinkFilterInfo],
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Link.Audit"))],
+)
 async def get_filters(link: Link = Depends(dep_link)) -> List[dict]:
     """
     Return all filters available for a given link.
@@ -253,6 +258,7 @@ async def stop_capture(link: Link = Depends(dep_link)) -> None:
 
 @router.post(
     "/{link_id}/capture/wireshark/restart",
+    response_model=schemas.LinkWiresharkRestart,
     status_code=status.HTTP_200_OK,
     dependencies=[Depends(has_privilege("Link.Capture"))],
 )
@@ -405,7 +411,12 @@ async def web_wireshark_websocket(
             pass
 
 
-@router.get("/{link_id}/markers", dependencies=[Depends(has_privilege("Link.Audit"))])
+@router.get(
+    "/{link_id}/markers",
+    response_model=Dict[str, schemas.Marker],
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Link.Audit"))],
+)
 async def get_markers(link: Link = Depends(dep_link)) -> dict:
     """
     Return all traffic-insight markers configured on this link.
@@ -417,7 +428,11 @@ async def get_markers(link: Link = Depends(dep_link)) -> dict:
 
 
 @router.post(
-    "/{link_id}/markers", status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("Link.Modify"))]
+    "/{link_id}/markers",
+    response_model=schemas.Marker,
+    response_model_exclude_unset=True,
+    status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(has_privilege("Link.Modify"))],
 )
 async def create_marker(marker_data: schemas.MarkerCreate, link: Link = Depends(dep_link)) -> dict:
     """
@@ -462,7 +477,12 @@ async def delete_marker(marker_name: str, link: Link = Depends(dep_link)) -> Non
     await link.stop_marker(marker_name)
 
 
-@router.put("/{link_id}/markers/{marker_name}", dependencies=[Depends(has_privilege("Link.Modify"))])
+@router.put(
+    "/{link_id}/markers/{marker_name}",
+    response_model=schemas.Marker,
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Link.Modify"))],
+)
 async def update_marker(marker_name: str, marker_data: schemas.MarkerUpdate, link: Link = Depends(dep_link)) -> dict:
     """
     Update a traffic-insight marker (change BPF, tag, or enabled).

gns3server/api/routes/controller/nodes.py

@@ -447,7 +447,11 @@ async def get_node_links(node: Node = Depends(dep_node)) -> List[schemas.Link]:
     return links
 
 
-@router.get("/{node_id}/dynamips/auto_idlepc", dependencies=[Depends(has_privilege("Node.Audit"))])
+@router.get(
+    "/{node_id}/dynamips/auto_idlepc",
+    response_model=schemas.IdlePC,
+    dependencies=[Depends(has_privilege("Node.Audit"))],
+)
 async def auto_idlepc(node: Node = Depends(dep_node)) -> dict:
     """
     Compute an Idle-PC value for a Dynamips node

gns3server/api/routes/controller/projects.py

@@ -31,7 +31,7 @@ import gns3server.utils.zipfile_zstd as zipfile
 
 log = logging.getLogger()
 
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 from uuid import UUID
 
 from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
@@ -199,7 +199,9 @@ async def delete_project(
     await rbac_repo.delete_all_ace_starting_with_path(f"/projects/{project.id}")
 
 
-@router.get("/{project_id}/stats", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+    "/{project_id}/stats", response_model=schemas.ProjectStats, dependencies=[Depends(has_privilege("Project.Audit"))]
+)
 def get_project_stats(project: Project = Depends(dep_project)) -> dict:
     """
     Return a project statistics.
@@ -210,7 +212,12 @@ def get_project_stats(project: Project = Depends(dep_project)) -> dict:
     return project.stats()
 
 
-@router.get("/{project_id}/markers", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+    "/{project_id}/markers",
+    response_model=Dict[str, schemas.ProjectMarker],
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Project.Audit"))],
+)
 def get_project_markers(project: Project = Depends(dep_project)) -> dict:
     """
     Return all traffic-insight markers across every link in the project.
@@ -241,6 +248,7 @@ async def _replay_response(awaitable):
 
 @router.get(
     "/{project_id}/markers/tags/{tag}/replay/range",
+    response_model=schemas.ReplayTimeline,
     dependencies=[Depends(has_privilege("Project.Audit"))],
 )
 async def replay_tag_range(
@@ -281,6 +289,7 @@ async def replay_tag_range(
 
 @router.get(
     "/{project_id}/markers/tags/{tag}/replay/frames",
+    response_model=schemas.ReplayFrames,
     dependencies=[Depends(has_privilege("Project.Audit"))],
 )
 async def replay_tag_frames(
@@ -321,6 +330,8 @@ async def replay_tag_frames(
 
 @router.get(
     "/{project_id}/markers/tags/{tag}/replay/frame/detail",
+    response_model=schemas.ReplayFrameDetail,
+    response_model_exclude_unset=True,
     dependencies=[Depends(has_privilege("Project.Audit"))],
 )
 async def replay_tag_frame_detail(
@@ -359,7 +370,12 @@ async def replay_tag_frame_detail(
 # ---------------------------------------------------------------------------
 
 
-@router.get("/{project_id}/marker-definitions", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+    "/{project_id}/marker-definitions",
+    response_model=Dict[str, schemas.MarkerDefinitionWithLinks],
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Project.Audit"))],
+)
 def get_marker_definitions(project: Project = Depends(dep_project)) -> dict:
     """
     Return all project-level marker definitions with their bound link IDs.
@@ -381,6 +397,8 @@ def get_marker_definitions(project: Project = Depends(dep_project)) -> dict:
 
 @router.post(
     "/{project_id}/marker-definitions",
+    response_model=schemas.MarkerDefinition,
+    response_model_exclude_unset=True,
     status_code=status.HTTP_201_CREATED,
     dependencies=[Depends(has_privilege("Project.Modify"))],
 )
@@ -408,7 +426,12 @@ async def create_marker_definition(
     return project.marker_definitions.get(name, {})
 
 
-@router.put("/{project_id}/marker-definitions/{def_name}", dependencies=[Depends(has_privilege("Project.Modify"))])
+@router.put(
+    "/{project_id}/marker-definitions/{def_name}",
+    response_model=schemas.MarkerDefinition,
+    response_model_exclude_unset=True,
+    dependencies=[Depends(has_privilege("Project.Modify"))],
+)
 async def update_marker_definition(
     def_name: str, def_data: schemas.MarkerDefinitionCreate, project: Project = Depends(dep_project)
 ) -> dict:

gns3server/api/routes/controller/symbols.py

@@ -21,7 +21,7 @@ API routes for symbols.
 
 import logging
 import os
-from typing import List
+from typing import Dict, List
 
 from fastapi import APIRouter, Depends, Request, Response, status
 from fastapi.responses import FileResponse
@@ -39,7 +39,7 @@ log = logging.getLogger(__name__)
 router = APIRouter()
 
 
-@router.get("", dependencies=[Depends(has_privilege("Symbol.Audit"))])
+@router.get("", response_model=List[schemas.SymbolInfo], dependencies=[Depends(has_privilege("Symbol.Audit"))])
 def get_symbols() -> List[dict]:
     """
     Return all symbols.
@@ -78,6 +78,7 @@ async def get_symbol(symbol_id: str, request: Request) -> Response:
 
 @router.get(
     "/{symbol_id:path}/dimensions",
+    response_model=schemas.SymbolDimensions,
     responses={404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"}},
     dependencies=[Depends(has_privilege("Symbol.Audit"))],
 )
@@ -97,7 +98,11 @@ async def get_symbol_dimensions(symbol_id: str) -> dict:
         raise ControllerNotFoundError(f"Could not get symbol file: {e}")
 
 
-@router.get("/default_symbols", dependencies=[Depends(has_privilege("Symbol.Audit"))])
+@router.get(
+    "/default_symbols",
+    response_model=Dict[str, Dict[str, str]],
+    dependencies=[Depends(has_privilege("Symbol.Audit"))],
+)
 def get_default_symbols() -> dict:
     """
     Return all default symbols.