mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-09-30 07:40:12 +03:00
Merge pull request #2919 from markparonyan/mypy-api-routes-compute-3
fix(typing): resolve mypy errors in gns3server.api.routes.compute
This commit is contained in:
commit
87f469b299
@ -23,7 +23,7 @@ import os
|
||||
from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from fastapi.responses import StreamingResponse
|
||||
from typing import Union
|
||||
from typing import Any, Union
|
||||
from uuid import UUID
|
||||
|
||||
from gns3server import schemas
|
||||
@ -32,7 +32,9 @@ from gns3server.compute.iou.iou_vm import IOUVM
|
||||
|
||||
from .dependencies.authentication import compute_authentication, ws_compute_authentication
|
||||
|
||||
responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}}
|
||||
responses: dict[int | str, dict[str, Any]] = {
|
||||
404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}
|
||||
}
|
||||
|
||||
router = APIRouter(responses=responses)
|
||||
|
||||
@ -60,18 +62,18 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
|
||||
"""
|
||||
|
||||
iou = IOU.instance()
|
||||
node_data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
vm = await iou.create_node(
|
||||
node_data.pop("name"),
|
||||
data.pop("name"),
|
||||
str(project_id),
|
||||
node_data.get("node_id"),
|
||||
application_id=node_data.get("application_id"),
|
||||
path=node_data.get("path"),
|
||||
console=node_data.get("console"),
|
||||
console_type=node_data.get("console_type", "telnet"),
|
||||
data.get("node_id"),
|
||||
application_id=data.get("application_id"),
|
||||
path=data.get("path"),
|
||||
console=data.get("console"),
|
||||
console_type=data.get("console_type", "telnet"),
|
||||
)
|
||||
|
||||
for name, value in node_data.items():
|
||||
for name, value in data.items():
|
||||
if hasattr(vm, name) and getattr(vm, name) != value:
|
||||
if name == "application_id":
|
||||
continue # we must ignore this to avoid overwriting the application_id allocated by the controller
|
||||
@ -79,7 +81,7 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
|
||||
continue
|
||||
if name == "private_config_content" and (vm.private_config_content and len(vm.private_config_content) > 0):
|
||||
continue
|
||||
if node_data.get("use_default_iou_values") and (name == "ram" or name == "nvram"):
|
||||
if data.get("use_default_iou_values") and (name == "ram" or name == "nvram"):
|
||||
continue
|
||||
setattr(vm, name, value)
|
||||
return vm.asdict()
|
||||
@ -100,8 +102,8 @@ async def update_iou_node(node_data: schemas.IOUUpdate, node: IOUVM = Depends(de
|
||||
Update an IOU node.
|
||||
"""
|
||||
|
||||
node_data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
for name, value in node_data.items():
|
||||
data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
for name, value in data.items():
|
||||
if hasattr(node, name) and getattr(node, name) != value:
|
||||
if name == "application_id":
|
||||
continue # we must ignore this to avoid overwriting the application_id allocated by the IOU manager
|
||||
@ -147,8 +149,8 @@ async def start_iou_node(start_data: schemas.IOUStart, node: IOUVM = Depends(dep
|
||||
Start an IOU node.
|
||||
"""
|
||||
|
||||
start_data = jsonable_encoder(start_data, exclude_unset=True)
|
||||
for name, value in start_data.items():
|
||||
start_values = jsonable_encoder(start_data, exclude_unset=True)
|
||||
for name, value in start_values.items():
|
||||
if hasattr(node, name) and getattr(node, name) != value:
|
||||
setattr(node, name, value)
|
||||
|
||||
@ -223,7 +225,7 @@ async def update_iou_node_nio(
|
||||
|
||||
nio = node.get_nio(adapter_number, port_number)
|
||||
nio.filters.clear()
|
||||
if nio_data.filters:
|
||||
if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
|
||||
nio.filters = nio_data.filters
|
||||
# NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers.
|
||||
nio.markers = getattr(nio_data, "markers", None) or {}
|
||||
|
||||
@ -23,14 +23,16 @@ import os
|
||||
from fastapi import APIRouter, Depends, Path, status, HTTPException
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from fastapi.responses import StreamingResponse
|
||||
from typing import Union
|
||||
from typing import Any, Union
|
||||
from uuid import UUID
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.compute.builtin import Builtin
|
||||
from gns3server.compute.builtin.nodes.nat import Nat
|
||||
|
||||
responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"}}
|
||||
responses: dict[int | str, dict[str, Any]] = {
|
||||
404: {"model": schemas.ErrorMessage, "description": "Could not find project or NAT node"}
|
||||
}
|
||||
|
||||
router = APIRouter(responses=responses)
|
||||
|
||||
@ -57,16 +59,16 @@ async def create_nat_node(project_id: UUID, node_data: schemas.NATCreate) -> sch
|
||||
"""
|
||||
|
||||
builtin_manager = Builtin.instance()
|
||||
node_data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
node = await builtin_manager.create_node(
|
||||
node_data.pop("name"),
|
||||
data.pop("name"),
|
||||
str(project_id),
|
||||
node_data.get("node_id"),
|
||||
data.get("node_id"),
|
||||
node_type="nat",
|
||||
ports=node_data.get("ports_mapping"),
|
||||
ports=data.get("ports_mapping"),
|
||||
)
|
||||
|
||||
node.usage = node_data.get("usage", "")
|
||||
node.usage = data.get("usage", "")
|
||||
return node.asdict()
|
||||
|
||||
|
||||
@ -85,8 +87,8 @@ async def update_nat_node(node_data: schemas.NATUpdate, node: Nat = Depends(dep_
|
||||
Update a NAT node.
|
||||
"""
|
||||
|
||||
node_data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
for name, value in node_data.items():
|
||||
data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
for name, value in data.items():
|
||||
if hasattr(node, name) and getattr(node, name) != value:
|
||||
setattr(node, name, value)
|
||||
node.updated()
|
||||
@ -170,7 +172,7 @@ async def update_nat_node_nio(
|
||||
|
||||
nio = node.get_nio(port_number)
|
||||
nio.filters.clear()
|
||||
if nio_data.filters:
|
||||
if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
|
||||
nio.filters = nio_data.filters
|
||||
await node.update_nio(port_number, nio)
|
||||
return nio.asdict()
|
||||
|
||||
@ -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.iou_nodes", # 15
|
||||
"gns3server.api.routes.compute.nat_nodes", # 10
|
||||
"gns3server.api.routes.compute.projects", # 7
|
||||
"gns3server.api.routes.compute.virtualbox_nodes", # 20
|
||||
"gns3server.api.routes.compute.vmware_nodes", # 10
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user