diff --git a/gns3server/api/routes/compute/projects.py b/gns3server/api/routes/compute/projects.py index 7c1248cc5..729510055 100644 --- a/gns3server/api/routes/compute/projects.py +++ b/gns3server/api/routes/compute/projects.py @@ -44,7 +44,7 @@ from gns3server import schemas router = APIRouter() # How many clients have subscribed to notifications -_notifications_listening = {} +_notifications_listening: dict[str, int] = {} def dep_project(project_id: UUID) -> Project: @@ -74,12 +74,12 @@ def create_compute_project(project_data: schemas.ProjectCreate) -> schemas.Proje """ pm = ProjectManager.instance() - project_data = jsonable_encoder(project_data, exclude_unset=True) + data = jsonable_encoder(project_data, exclude_unset=True) project = pm.create_project( - name=project_data.get("name"), - path=project_data.get("path"), - project_id=project_data.get("project_id"), - variables=project_data.get("variables", None), + name=data.get("name"), + path=data.get("path"), + project_id=data.get("project_id"), + variables=data.get("variables", None), ) return project.asdict() @@ -261,7 +261,7 @@ async def create_batch_nios( # This is what makes builtin L2 nodes (ethernet_switch/hub/cloud/nat) # start their uBridge concurrently during project open instead of one # at a time (~0.5s each for fork + socket connect). - per_node = {} + per_node: dict[str, list[schemas.BatchNIOEntry]] = {} for entry in batch.nios: per_node.setdefault(entry.node_id, []).append(entry) @@ -308,7 +308,7 @@ async def update_batch_nios( # Group entries by node so that different nodes' uBridge processes are # updated in parallel (each node has its own AF_UNIX socket). Within a # node entries are serial to respect the per-node uBridge command lock. - per_node = {} + per_node: dict[str, list[schemas.BatchNIOEntry]] = {} for entry in batch.nios: per_node.setdefault(entry.node_id, []).append(entry) diff --git a/gns3server/api/routes/compute/virtualbox_nodes.py b/gns3server/api/routes/compute/virtualbox_nodes.py index ec040d3e4..9f137d066 100644 --- a/gns3server/api/routes/compute/virtualbox_nodes.py +++ b/gns3server/api/routes/compute/virtualbox_nodes.py @@ -24,7 +24,7 @@ from fastapi import APIRouter, WebSocket, Depends, Path, status from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse from uuid import UUID -from typing import Union +from typing import Any, Union from gns3server import schemas from gns3server.compute.virtualbox import VirtualBox @@ -33,7 +33,9 @@ from gns3server.compute.virtualbox.virtualbox_vm import VirtualBoxVM from .dependencies.authentication import compute_authentication, ws_compute_authentication -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or VirtualBox node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or VirtualBox node"} +} router = APIRouter(responses=responses, deprecated=True) @@ -61,24 +63,24 @@ async def create_virtualbox_node(project_id: UUID, node_data: schemas.VirtualBox """ vbox_manager = VirtualBox.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) vm = await vbox_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), - node_data.pop("vmname"), - linked_clone=node_data.pop("linked_clone", False), - console=node_data.get("console", None), - console_type=node_data.get("console_type", "telnet"), - adapters=node_data.get("adapters", 0), + data.get("node_id"), + data.pop("vmname"), + linked_clone=data.pop("linked_clone", False), + console=data.get("console", None), + console_type=data.get("console_type", "telnet"), + adapters=data.get("adapters", 0), ) - if "ram" in node_data: - ram = node_data.pop("ram") + if "ram" in data: + ram = data.pop("ram") if ram != vm.ram: await vm.set_ram(ram) - for name, value in node_data.items(): + for name, value in data.items(): if name != "node_id": if hasattr(vm, name) and getattr(vm, name) != value: setattr(vm, name, value) @@ -103,10 +105,10 @@ async def update_virtualbox_node( Update a VirtualBox node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data: - name = node_data.pop("name") - vmname = node_data.pop("vmname", None) + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data: + name = data.pop("name") + vmname = data.pop("vmname", None) if name != node.name: oldname = node.name node.name = name @@ -118,20 +120,20 @@ async def update_virtualbox_node( node.updated() raise e - if "adapters" in node_data: - adapters = node_data.pop("adapters") + if "adapters" in data: + adapters = data.pop("adapters") if adapters != node.adapters: await node.set_adapters(adapters) - if "ram" in node_data: - ram = node_data.pop("ram") + if "ram" in data: + ram = data.pop("ram") if ram != node.ram: await node.set_ram(ram) # update the console first to avoid issue if updating console type - node.console = node_data.pop("console", node.console) + node.console = data.pop("console", node.console) - for name, value in node_data.items(): + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: setattr(node, name, value) diff --git a/pyproject.toml b/pyproject.toml index 5371f8745..bb1f95c75 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -294,8 +294,6 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] # Baseline: modules with existing type errors (error count at baseline time). # Remove modules from this list one small PR at a time. Never add new ones. module = [ - "gns3server.api.routes.compute.projects", # 7 - "gns3server.api.routes.compute.virtualbox_nodes", # 20 "gns3server.api.routes.compute.vmware_nodes", # 10 "gns3server.api.routes.compute.vpcs_nodes", # 9 "gns3server.api.routes.controller.acl", # 5