NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-09-30 07:40:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2921 from markparonyan/mypy-api-routes

Browse Source 
fix(typing): resolve mypy errors in gns3server.api.routes
This commit is contained in:
Jeremy Grossmann 2026-09-29 13:25:46 +02:00 committed by GitHub
commit 7e5f5486eb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 36 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
gns3server/api/routes/compute/vmware_nodes.py
View File

@ -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.vmware import VMware
@ -32,7 +32,9 @@ from gns3server.compute.vmware.vmware_vm import VMwareVM
from .dependencies.authentication import compute_authentication, ws_compute_authentication
responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or VMware node"}}
responses: dict[int | str, dict[str, Any]] = {
404: {"model": schemas.ErrorMessage, "description": "Could not find project or VMware node"}
}
router = APIRouter(responses=responses, deprecated=True)
@ -60,18 +62,18 @@ async def create_vmware_node(project_id: UUID, node_data: schemas.VMwareCreate)
"""
vmware_manager = VMware.instance()
node_data = jsonable_encoder(node_data, exclude_unset=True)
data = jsonable_encoder(node_data, exclude_unset=True)
vm = await vmware_manager.create_node(
node_data.pop("name"),
data.pop("name"),
str(project_id),
node_data.get("node_id"),
node_data.pop("vmx_path"),
linked_clone=node_data.pop("linked_clone"),
console=node_data.get("console", None),
console_type=node_data.get("console_type", "telnet"),
data.get("node_id"),
data.pop("vmx_path"),
linked_clone=data.pop("linked_clone"),
console=data.get("console", None),
console_type=data.get("console_type", "telnet"),
)
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)
@ -94,10 +96,10 @@ async def update_vmware_node(node_data: schemas.VMwareUpdate, node: VMwareVM = D
Update a VMware node.
"""
node_data = jsonable_encoder(node_data, exclude_unset=True)
data = jsonable_encoder(node_data, exclude_unset=True)
# update the console first to avoid issue if updating console type
node.console = node_data.pop("console", node.console)
for name, value in node_data.items():
node.console = data.pop("console", node.console)
for name, value in data.items():
if hasattr(node, name) and getattr(node, name) != value:
setattr(node, name, value)

26
gns3server/api/routes/compute/vpcs_nodes.py
View File

@ -23,7 +23,7 @@ import os
from fastapi import APIRouter, WebSocket, Depends, Body, 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
@ -31,7 +31,9 @@ from gns3server.compute.vpcs import VPCS
from gns3server.compute.vpcs.vpcs_vm import VPCSVM
from .dependencies.authentication import compute_authentication, ws_compute_authentication
responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or VMware node"}}
responses: dict[int | str, dict[str, Any]] = {
404: {"model": schemas.ErrorMessage, "description": "Could not find project or VMware node"}
}
router = APIRouter(responses=responses)
@ -59,14 +61,14 @@ async def create_vpcs_node(project_id: UUID, node_data: schemas.VPCSCreate) -> s
"""
vpcs = VPCS.instance()
node_data = jsonable_encoder(node_data, exclude_unset=True)
data = jsonable_encoder(node_data, exclude_unset=True)
vm = await vpcs.create_node(
node_data["name"],
data["name"],
str(project_id),
node_data.get("node_id"),
console=node_data.get("console"),
console_type=node_data.get("console_type", "telnet"),
startup_script=node_data.get("startup_script"),
data.get("node_id"),
console=data.get("console"),
console_type=data.get("console_type", "telnet"),
startup_script=data.get("startup_script"),
)
return vm.asdict()
@ -87,14 +89,14 @@ async def update_vpcs_node(node_data: schemas.VPCSUpdate, node: VPCSVM = Depends
Update a VPCS node.
"""
node_data = jsonable_encoder(node_data, exclude_unset=True)
name = node_data.get("name", node.name)
data = jsonable_encoder(node_data, exclude_unset=True)
name = data.get("name", node.name)
if node.name != name:
node.name = name
console = node_data.get("console", node.console)
console = data.get("console", node.console)
if node.console != console:
node.console = console
console_type = node_data.get("console_type", node.console_type)
console_type = data.get("console_type", node.console_type)
if node.console_type != console_type:
node.console_type = console_type
node.updated()

13
gns3server/api/routes/controller/acl.py
View File

@ -25,13 +25,14 @@ from fastapi import APIRouter, Depends, Request, status
from fastapi.routing import APIRoute, _IncludedRouter
from starlette.routing import BaseRoute, Mount
from uuid import UUID
from typing import Iterator, List, Sequence
from typing import Iterator, List, Optional, Sequence
from gns3server import schemas
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerNotFoundError
from gns3server.controller import Controller
import gns3server.db.models as models
from gns3server.db.repositories.users import UsersRepository
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.db.repositories.images import ImagesRepository
@ -96,7 +97,7 @@ async def endpoints(
endpoints = [{"endpoint": "/", "name": "All endpoints", "endpoint_type": "root"}]
def add_to_endpoints(endpoint: str, name: str, endpoint_type: str) -> None:
if endpoint not in endpoints:
if not any(e["endpoint"] == endpoint for e in endpoints):
endpoints.append({"endpoint": endpoint, "name": name, "endpoint_type": endpoint_type})
# projects
@ -175,7 +176,7 @@ async def endpoints(
@router.get("", response_model=List[schemas.ACE], dependencies=[Depends(has_privilege("ACE.Audit"))])
async def get_aces(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[schemas.ACE]:
async def get_aces(rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))) -> List[models.ACE]:
"""
Get all ACL entries.
@ -193,7 +194,7 @@ async def get_aces(rbac_repo: RbacRepository = Depends(get_repository(RbacReposi
)
async def create_ace(
request: Request, ace_create: schemas.ACECreate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
) -> schemas.ACE:
) -> models.ACE:
"""
Create a new ACL entry.
@ -224,7 +225,7 @@ async def create_ace(
async def get_ace(
ace_id: UUID,
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
) -> schemas.ACE:
) -> models.ACE:
"""
Get an ACL entry.
@ -240,7 +241,7 @@ async def get_ace(
@router.put("/{ace_id}", response_model=schemas.ACE, dependencies=[Depends(has_privilege("ACE.Modify"))])
async def update_ace(
ace_id: UUID, ace_update: schemas.ACEUpdate, rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
) -> schemas.ACE:
) -> Optional[models.ACE]:
"""
Update an ACL entry.

3
pyproject.toml
View File

@ -294,9 +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.vmware_nodes", # 10
"gns3server.api.routes.compute.vpcs_nodes", # 9
"gns3server.api.routes.controller.acl", # 5
"gns3server.api.routes.controller.computes", # 6
"gns3server.api.routes.controller.controller", # 10
"gns3server.api.routes.controller.dependencies.authentication", # 11