Merge pull request #2923 from markparonyan/mypy-api-routes-controller-2

fix(typing): resolve mypy errors in gns3server.api.routes.controller

gns3server/api/routes/controller/nodes.py

@@ -27,7 +27,7 @@ from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Request,
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
 from fastapi.routing import APIRoute
-from typing import List, Callable, Optional
+from typing import Any, List, Callable, Optional
 from uuid import UUID
 
 from gns3server.controller import Controller
@@ -47,7 +47,7 @@ import logging
 
 log = logging.getLogger(__name__)
 
-node_locks = {}
+node_locks: dict[str, dict[str, Any]] = {}
 
 
 class NodeConcurrency(APIRoute):
@@ -84,7 +84,9 @@ class NodeConcurrency(APIRoute):
         return custom_route_handler
 
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or node"}
+}
 
 router = APIRouter(route_class=NodeConcurrency, responses=responses)
 
@@ -127,7 +129,7 @@ def _check_node_type(node: Node, *required_types: str) -> None:
     },
     dependencies=[Depends(has_privilege("Node.Allocate"))],
 )
-async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node:
+async def create_node(node_create: schemas.NodeCreate, project: Project = Depends(dep_project)) -> schemas.Node:
     """
     Create a new node.
 
@@ -135,8 +137,8 @@ async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(
     """
 
     controller = Controller.instance()
-    compute = controller.get_compute(str(node_data.compute_id))
+    compute = controller.get_compute(str(node_create.compute_id))
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    node_data = jsonable_encoder(node_create, exclude_unset=True)
     node = await project.add_node(compute, node_data.pop("name"), node_data.pop("node_id", None), **node_data)
     return node.asdict()
 
@@ -274,14 +276,14 @@ async def get_node(node: Node = Depends(dep_node)) -> schemas.Node:
     response_model_exclude_unset=True,
     dependencies=[Depends(has_privilege("Node.Modify"))],
 )
-async def update_node(node_data: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node:
+async def update_node(node_update: schemas.NodeUpdate, node: Node = Depends(dep_node)) -> schemas.Node:
     """
     Update a node.
 
     Required privilege: Node.Modify
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    node_data = jsonable_encoder(node_update, exclude_unset=True)
 
     # Ignore these because we only use them when creating a node
     node_data.pop("node_id", None)
@@ -654,9 +656,8 @@ async def ws_console(
         return
 
     compute = node.compute
-    log.info(
+    client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown"
-        f"New client {websocket.client.host}:{websocket.client.port} has connected to controller console WebSocket"
+    log.info(f"New client {client} has connected to controller console WebSocket")
-    )
 
     compute_host = compute.host
     try:
@@ -690,9 +691,7 @@ async def ws_console(
                     await ws_console_compute.send_bytes(msg["bytes"])
         except WebSocketDisconnect:
             pass
-        log.info(
+        log.info(f"Client {client} has disconnected from controller console WebSocket")
-            f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller console WebSocket"
-        )
 
     async def ws_send(ws_console_compute):
         """
@@ -709,10 +708,7 @@ async def ws_console(
                     break
         except WebSocketDisconnect:
             # the client disconnected while the compute was still streaming console output
-            log.info(
+            log.info(f"Client {client} has disconnected from controller console WebSocket")
-                f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-                f" console WebSocket"
-            )
 
     try:
         # forward WebSocket data in both directions between the client and the compute console WebSocket
@@ -763,9 +759,8 @@ async def vnc_console(
         return
 
     compute = node.compute
-    log.info(
+    client = f"{websocket.client.host}:{websocket.client.port}" if websocket.client else "unknown"
-        f"New client {websocket.client.host}:{websocket.client.port} has connected to controller VNC console WebSocket"
+    log.info(f"New client {client} has connected to controller VNC console WebSocket")
-    )
 
     compute_host = compute.host
     try:
@@ -796,10 +791,7 @@ async def vnc_console(
                     await vnc_console_compute.send_bytes(data)
         except WebSocketDisconnect:
             pass
-        log.info(
+        log.info(f"Client {client} has disconnected from controller VNC console WebSocket")
-            f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-            f" VNC console WebSocket"
-        )
 
     async def vnc_send(vnc_console_compute):
         """
@@ -814,10 +806,7 @@ async def vnc_console(
                     break
         except WebSocketDisconnect:
             # the client disconnected while the compute was still streaming VNC console output
-            log.info(
+            log.info(f"Client {client} has disconnected from controller VNC console WebSocket")
-                f"Client {websocket.client.host}:{websocket.client.port} has disconnected from controller"
-                f" VNC console WebSocket"
-            )
 
     try:
         # forward WebSocket data in both directions between the client and the compute VNC console WebSocket

gns3server/api/routes/controller/privileges.py

@@ -16,6 +16,7 @@ API route for privileges
 """
 
 from typing import List
+import gns3server.db.models as models
 from gns3server.db.repositories.rbac import RbacRepository
 from .dependencies.database import get_repository
 from fastapi import APIRouter, Depends
@@ -33,7 +34,7 @@ router = APIRouter()
 )
 async def get_privileges(
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
-) -> List[schemas.Privilege]:
+) -> List[models.Privilege]:
     """
     Get all privileges.
 

gns3server/api/routes/controller/roles.py

@@ -21,7 +21,7 @@ API routes for roles.
 
 from fastapi import APIRouter, Depends, status
 from uuid import UUID
-from typing import List
+from typing import List, Optional
 
 from gns3server import schemas
 from gns3server.controller.controller_error import (
@@ -31,6 +31,7 @@ from gns3server.controller.controller_error import (
     ControllerForbiddenError,
 )
 
+import gns3server.db.models as models
 from gns3server.db.repositories.rbac import RbacRepository
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege
@@ -43,7 +44,7 @@ router = APIRouter()
 
 
 @router.get("", response_model=List[schemas.Role], dependencies=[Depends(has_privilege("Role.Audit"))])
-async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[schemas.Role]:
+async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[models.Role]:
     """
     Get all roles.
 
@@ -61,7 +62,7 @@ async def get_roles(rbac_repo: RbacRepository = Depends(get_repository(RbacRepos
 )
 async def create_role(
     role_create: schemas.RoleCreate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> schemas.Role:
+) -> Optional[models.Role]:
     """
     Create a new role.
 
@@ -78,7 +79,7 @@ async def create_role(
 async def get_role(
     role_id: UUID,
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
-) -> schemas.Role:
+) -> models.Role:
     """
     Get a role.
 
@@ -94,7 +95,7 @@ async def get_role(
 @router.put("/{role_id}", response_model=schemas.Role, dependencies=[Depends(has_privilege("Role.Modify"))])
 async def update_role(
     role_id: UUID, role_update: schemas.RoleUpdate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> schemas.Role:
+) -> Optional[models.Role]:
     """
     Update a role.
 
@@ -142,7 +143,7 @@ async def delete_role(
 )
 async def get_role_privileges(
     role_id: UUID, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
-) -> List[schemas.Privilege]:
+) -> List[models.Privilege]:
     """
     Get all role privileges.
 

gns3server/api/routes/controller/snapshots.py

@@ -24,7 +24,7 @@ import logging
 log = logging.getLogger()
 
 from fastapi import APIRouter, Depends, status
-from typing import List
+from typing import Any, List
 from uuid import UUID
 
 from gns3server.controller.project import Project
@@ -35,7 +35,9 @@ from gns3server.controller import Controller
 from .dependencies.database import get_repository
 from .dependencies.rbac import has_privilege
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or snapshot"}
+}
 
 router = APIRouter(responses=responses)
 

pyproject.toml

@@ -325,12 +325,8 @@ module = [
   "gns3server.api.routes.controller.groups",  # 6
   "gns3server.api.routes.controller.images",  # 9
   "gns3server.api.routes.controller.llm_model_configs",  # 32
-  "gns3server.api.routes.controller.nodes",  # 13
   "gns3server.api.routes.controller.pools",  # 8
-  "gns3server.api.routes.controller.privileges",  # 1
   "gns3server.api.routes.controller.projects",  # 16
-  "gns3server.api.routes.controller.roles",  # 5
-  "gns3server.api.routes.controller.snapshots",  # 1
   "gns3server.api.routes.controller.templates",  # 10
   "gns3server.api.routes.controller.users",  # 14
   "gns3server.api.server",  # 2