refactor: derive update schemas with PartialUpdateModel

This commit is contained in:
Mark Paronyan 2026-10-06 16:11:38 +03:00
parent 65c858d455
commit d9678ce98c
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
34 changed files with 276 additions and 70 deletions

View File

@ -105,8 +105,6 @@ async def update_iou_node(node_data: schemas.IOUUpdate, node: IOUVM = Depends(de
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
setattr(node, name, value)
if node.use_default_iou_values:

View File

@ -284,11 +284,6 @@ async def update_node(node_update: schemas.NodeUpdate, node: Node = Depends(dep_
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)
node_data.pop("node_type", None)
node_data.pop("compute_id", None)
await node.update(**node_data)
return node.asdict()

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel
from ..common import NodeStatus
from ..update import PartialUpdateModel
class ATMSwitchBase(BaseModel):
@ -41,11 +42,13 @@ class ATMSwitchCreate(ATMSwitchBase):
node_id: Optional[UUID] = None
class ATMSwitchUpdate(ATMSwitchBase):
class ATMSwitchUpdate(PartialUpdateModel, ATMSwitchBase):
"""
Properties to update an ATM switch node.
"""
update_excluded_fields = ("node_id",)
name: Optional[str] = None
node_id: Optional[UUID] = None

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import NodeStatus
from ..update import PartialUpdateModel
class HostInterfaceType(str, Enum):
@ -148,12 +149,14 @@ class CloudCreate(CloudBase):
name: str
class CloudUpdate(CloudBase):
class CloudUpdate(PartialUpdateModel, CloudBase):
"""
Properties to update a cloud node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
class Cloud(CloudBase):

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel, Field, field_validator
from ..common import AuxType, ConsoleType, CustomAdapter, ExtraConfig, NodeStatus
from ..update import PartialUpdateModel
class DockerBase(BaseModel):
@ -91,12 +92,15 @@ class DockerCreate(DockerBase):
)
class DockerUpdate(DockerBase):
class DockerUpdate(PartialUpdateModel, DockerBase):
"""
Properties to update a Docker node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
image: Optional[str] = Field(None, description="Docker image name")
class Docker(DockerBase):

View File

@ -22,6 +22,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import NodeStatus
from ..update import PartialUpdateModel
class DynamipsPlatform(str, Enum):
@ -180,11 +181,13 @@ class DynamipsCreate(DynamipsBase):
ram: int = Field(..., gt=0, description="Amount of RAM in MB")
class DynamipsUpdate(DynamipsBase):
class DynamipsUpdate(PartialUpdateModel, DynamipsBase):
"""
Properties to update a Dynamips node.
"""
update_excluded_fields = ("node_id", "dynamips_id")
platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel
from ..common import NodeStatus
from ..update import PartialUpdateModel
class EthernetHubPort(BaseModel):
@ -46,12 +47,12 @@ class EthernetHubCreate(EthernetHubBase):
name: str
class EthernetHubUpdate(EthernetHubBase):
class EthernetHubUpdate(PartialUpdateModel, EthernetHubBase):
"""
Properties to update an Ethernet hub node.
"""
pass
update_excluded_fields = ("node_id",)
class EthernetHub(EthernetHubBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field, model_validator
from ..common import NodeStatus
from ..update import PartialUpdateModel
class EthernetSwitchPortType(str, Enum):
@ -83,12 +84,12 @@ class EthernetSwitchCreate(EthernetSwitchBase):
name: str
class EthernetSwitchUpdate(EthernetSwitchBase):
class EthernetSwitchUpdate(PartialUpdateModel, EthernetSwitchBase):
"""
Properties to update an Ethernet hub node.
"""
pass
update_excluded_fields = ("node_id",)
class EthernetSwitch(EthernetSwitchBase):

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel
from ..common import NodeStatus
from ..update import PartialUpdateModel
class FrameRelaySwitchBase(BaseModel):
@ -41,11 +42,13 @@ class FrameRelaySwitchCreate(FrameRelaySwitchBase):
node_id: Optional[UUID] = None
class FrameRelaySwitchUpdate(FrameRelaySwitchBase):
class FrameRelaySwitchUpdate(PartialUpdateModel, FrameRelaySwitchBase):
"""
Properties to update an Frame Relay node.
"""
update_excluded_fields = ("node_id",)
name: Optional[str] = None
node_id: Optional[UUID] = None

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import ConsoleType, NodeStatus
from ..update import PartialUpdateModel
class IOUBase(BaseModel):
@ -58,12 +59,16 @@ class IOUCreate(IOUBase):
application_id: int = Field(..., description="Application ID for running IOU executable")
class IOUUpdate(IOUBase):
class IOUUpdate(PartialUpdateModel, IOUBase):
"""
Properties to update an IOU node.
"""
pass
update_excluded_fields = ("node_id", "application_id")
name: Optional[str] = None
path: Optional[str] = Field(None, description="IOU executable path")
application_id: Optional[int] = Field(None, description="Application ID for running IOU executable")
class IOU(IOUBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import NodeStatus
from ..update import PartialUpdateModel
class HostInterfaceType(str, Enum):
@ -106,12 +107,14 @@ class NATCreate(NATBase):
name: str
class NATUpdate(NATBase):
class NATUpdate(PartialUpdateModel, NATBase):
"""
Properties to update a NAT node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
class NAT(NATBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
class QemuPlatform(str, Enum):
@ -227,12 +228,14 @@ class QemuCreate(QemuBase):
)
class QemuUpdate(QemuBase):
class QemuUpdate(PartialUpdateModel, QemuBase):
"""
Properties to update a Qemu node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
class Qemu(QemuBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
class VirtualBoxConsoleType(str, Enum):
@ -83,12 +84,15 @@ class VirtualBoxCreate(VirtualBoxBase):
vmname: str = Field(..., description="VirtualBox VM name (in VirtualBox itself)")
class VirtualBoxUpdate(VirtualBoxBase):
class VirtualBoxUpdate(PartialUpdateModel, VirtualBoxBase):
"""
Properties to update a VirtualBox node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
vmname: Optional[str] = None
class VirtualBox(VirtualBoxBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
class VMwareConsoleType(str, Enum):
@ -89,12 +90,16 @@ class VMwareCreate(VMwareBase):
linked_clone: bool = Field(..., description="Whether the VM is a linked clone or not")
class VMwareUpdate(VMwareBase):
class VMwareUpdate(PartialUpdateModel, VMwareBase):
"""
Properties to update a VMware node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
vmx_path: Optional[str] = None
linked_clone: Optional[bool] = None
class VMware(VMwareBase):

View File

@ -21,6 +21,7 @@ from uuid import UUID
from pydantic import BaseModel, Field
from ..common import NodeStatus
from ..update import PartialUpdateModel
class ConsoleType(str, Enum):
@ -54,12 +55,14 @@ class VPCSCreate(VPCSBase):
name: str
class VPCSUpdate(VPCSBase):
class VPCSUpdate(PartialUpdateModel, VPCSBase):
"""
Properties to update a VPCS node.
"""
pass
update_excluded_fields = ("node_id",)
name: Optional[str] = None
class VPCS(VPCSBase):

View File

@ -20,6 +20,7 @@ from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
from ..update import PartialUpdateModel
from .labels import Label
@ -73,7 +74,7 @@ class LinkCreate(LinkBase):
nodes: List[LinkNode] = Field(..., min_length=2, max_length=2)
class LinkUpdate(LinkBase):
class LinkUpdate(PartialUpdateModel, LinkBase):
pass

View File

@ -21,6 +21,7 @@ from uuid import UUID, uuid4
from pydantic import BaseModel, Field
from ..common import ConsoleType, CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
from .labels import Label
@ -169,12 +170,12 @@ class NodeCreate(NodeBase):
node_id: UUID = Field(default_factory=uuid4)
class NodeUpdate(NodeBase):
class NodeUpdate(PartialUpdateModel, NodeBase):
"""
Data to update a node.
"""
pass
update_excluded_fields = ("node_id", "compute_id", "node_type")
class Node(NodeBase):

View File

@ -21,6 +21,8 @@ from uuid import UUID
from pydantic import BaseModel, Field, HttpUrl
from ..update import PartialUpdateModel
class ProjectStatus(str, Enum):
"""
@ -82,12 +84,12 @@ class ProjectDuplicate(ProjectBase):
reset_mac_addresses: Optional[bool] = Field(False, description="Reset MAC addresses for this project")
class ProjectUpdate(ProjectBase):
class ProjectUpdate(PartialUpdateModel, ProjectBase):
"""
Properties for project update.
"""
pass
update_excluded_fields = ("project_id", "path")
class Project(ProjectBase):

View File

@ -70,7 +70,8 @@ class ACEUpdate(ACEBase):
Properties to update an ACE.
"""
pass
propagate: Optional[bool] = None
allowed: Optional[bool] = None
class ACE(DateTimeModelMixin, ACEBase):

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
from ...update import PartialUpdateModel
from ..base import DateTimeModelMixin
from ..nodes import NodeType
@ -98,7 +99,8 @@ class TemplateCreate(TemplateBase):
model_config = ConfigDict(extra="allow")
class TemplateUpdate(TemplateBase):
class TemplateUpdate(PartialUpdateModel, TemplateBase):
update_excluded_fields = ("template_id",)
model_config = ConfigDict(extra="allow")

View File

@ -21,6 +21,7 @@ from pydantic import Field
from gns3server.schemas.compute.cloud_nodes import CloudConsoleType, EthernetPort, TAPPort, UDPPort
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -35,5 +36,5 @@ class CloudTemplate(TemplateBase):
remote_console_http_path: Optional[str] = Field("/", description="Path of the remote web interface")
class CloudTemplateUpdate(CloudTemplate):
pass
class CloudTemplateUpdate(PartialUpdateModel, CloudTemplate):
update_excluded_fields = ("template_id",)

View File

@ -20,6 +20,7 @@ from typing import List, Optional
from pydantic import Field
from ...common import AuxType, ConsoleType, CustomAdapter, ExtraConfig
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -63,5 +64,5 @@ class DockerTemplate(DockerTemplateBase):
image: str = Field(..., description="Docker image name")
class DockerTemplateUpdate(DockerTemplateBase):
pass
class DockerTemplateUpdate(PartialUpdateModel, DockerTemplateBase):
update_excluded_fields = ("template_id",)

View File

@ -28,6 +28,7 @@ from gns3server.schemas.compute.dynamips_nodes import (
DynamipsWics,
)
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -73,7 +74,7 @@ class DynamipsTemplate(DynamipsTemplateBase):
image: str = Field(..., description="Path to the IOS image")
class C7200DynamipsTemplateUpdate(DynamipsTemplateBase):
class C7200DynamipsTemplateBase(DynamipsTemplateBase):
ram: Optional[int] = Field(512, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(512, gt=0, description="Amount of NVRAM in KB")
npe: Optional[DynamipsNPE] = Field(DynamipsNPE.npe_400, description="NPE model")
@ -81,31 +82,43 @@ class C7200DynamipsTemplateUpdate(DynamipsTemplateBase):
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C7200DynamipsTemplate(C7200DynamipsTemplateUpdate):
class C7200DynamipsTemplateUpdate(PartialUpdateModel, C7200DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C7200DynamipsTemplate(C7200DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
class C3725DynamipsTemplateUpdate(DynamipsTemplateBase):
class C3725DynamipsTemplateBase(DynamipsTemplateBase):
ram: Optional[int] = Field(128, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C3725DynamipsTemplate(C3725DynamipsTemplateUpdate):
class C3725DynamipsTemplateUpdate(PartialUpdateModel, C3725DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C3725DynamipsTemplate(C3725DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
class C3745DynamipsTemplateUpdate(DynamipsTemplateBase):
class C3745DynamipsTemplateBase(DynamipsTemplateBase):
ram: Optional[int] = Field(256, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C3745DynamipsTemplate(C3745DynamipsTemplateUpdate):
class C3745DynamipsTemplateUpdate(PartialUpdateModel, C3745DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C3745DynamipsTemplate(C3745DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
@ -116,7 +129,7 @@ class C3600ChassisType(str, Enum):
chassis_3660 = "3660"
class C3600DynamipsTemplateUpdate(DynamipsTemplateBase):
class C3600DynamipsTemplateBase(DynamipsTemplateBase):
chassis: Optional[C3600ChassisType] = Field(C3600ChassisType.chassis_3660, description="Chassis type")
ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
@ -124,19 +137,27 @@ class C3600DynamipsTemplateUpdate(DynamipsTemplateBase):
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C3600DynamipsTemplate(C3600DynamipsTemplateUpdate):
class C3600DynamipsTemplateUpdate(PartialUpdateModel, C3600DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C3600DynamipsTemplate(C3600DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
class C2691DynamipsTemplateUpdate(DynamipsTemplateBase):
class C2691DynamipsTemplateBase(DynamipsTemplateBase):
ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C2691DynamipsTemplate(C2691DynamipsTemplateUpdate):
class C2691DynamipsTemplateUpdate(PartialUpdateModel, C2691DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C2691DynamipsTemplate(C2691DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
@ -153,7 +174,7 @@ class C2600ChassisType(str, Enum):
chassis_2651xm = "2651XM"
class C2600DynamipsTemplateUpdate(DynamipsTemplateBase):
class C2600DynamipsTemplateBase(DynamipsTemplateBase):
chassis: Optional[C2600ChassisType] = Field(C2600ChassisType.chassis_2651xm, description="Chassis type")
ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
@ -161,7 +182,11 @@ class C2600DynamipsTemplateUpdate(DynamipsTemplateBase):
sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
class C2600DynamipsTemplate(C2600DynamipsTemplateUpdate):
class C2600DynamipsTemplateUpdate(PartialUpdateModel, C2600DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C2600DynamipsTemplate(C2600DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")
@ -174,7 +199,7 @@ class C1700ChassisType(str, Enum):
chassis_1760 = "1760"
class C1700DynamipsTemplateUpdate(DynamipsTemplateBase):
class C1700DynamipsTemplateBase(DynamipsTemplateBase):
chassis: Optional[C1700ChassisType] = Field(C1700ChassisType.chassis_1760, description="Chassis type")
ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB")
nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
@ -182,6 +207,10 @@ class C1700DynamipsTemplateUpdate(DynamipsTemplateBase):
sparsemem: Optional[bool] = Field(False, description="Sparse memory feature")
class C1700DynamipsTemplate(C1700DynamipsTemplateUpdate):
class C1700DynamipsTemplateUpdate(PartialUpdateModel, C1700DynamipsTemplateBase):
update_excluded_fields = ("template_id",)
class C1700DynamipsTemplate(C1700DynamipsTemplateBase):
platform: DynamipsPlatform = Field(..., description="Cisco router platform")
image: str = Field(..., description="Path to the IOS image")

View File

@ -20,6 +20,7 @@ from pydantic import Field
from gns3server.schemas.compute.ethernet_hub_nodes import EthernetHubPort
from ...update import PartialUpdateModel
from . import Category, TemplateBase
DEFAULT_PORTS = [
@ -41,5 +42,5 @@ class EthernetHubTemplate(TemplateBase):
ports_mapping: Optional[List[EthernetHubPort]] = Field(DEFAULT_PORTS, description="Ports")
class EthernetHubTemplateUpdate(EthernetHubTemplate):
pass
class EthernetHubTemplateUpdate(PartialUpdateModel, EthernetHubTemplate):
update_excluded_fields = ("template_id",)

View File

@ -22,6 +22,7 @@ from pydantic import Field
from gns3server.schemas.compute.ethernet_switch_nodes import EthernetSwitchPort
from ...update import PartialUpdateModel
from . import Category, TemplateBase
DEFAULT_PORTS = [
@ -53,5 +54,5 @@ class EthernetSwitchTemplate(TemplateBase):
console_type: Optional[ConsoleType] = Field(ConsoleType.none, description="Console type")
class EthernetSwitchTemplateUpdate(EthernetSwitchTemplate):
pass
class EthernetSwitchTemplateUpdate(PartialUpdateModel, EthernetSwitchTemplate):
update_excluded_fields = ("template_id",)

View File

@ -21,6 +21,7 @@ from pydantic import Field
from gns3server.schemas.compute.iou_nodes import ConsoleType
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -50,5 +51,5 @@ class IOUTemplate(IOUTemplateBase):
path: str = Field(..., description="Path of IOU executable")
class IOUTemplateUpdate(IOUTemplateBase):
pass
class IOUTemplateUpdate(PartialUpdateModel, IOUTemplateBase):
update_excluded_fields = ("template_id",)

View File

@ -30,6 +30,7 @@ from gns3server.schemas.compute.qemu_nodes import (
QemuProcessPriority,
)
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -102,5 +103,5 @@ class QemuTemplate(TemplateBase):
custom_adapters: Optional[List[CustomAdapter]] = Field(default_factory=list, description="Custom adapters")
class QemuTemplateUpdate(QemuTemplate):
pass
class QemuTemplateUpdate(PartialUpdateModel, QemuTemplate):
update_excluded_fields = ("template_id",)

View File

@ -25,6 +25,7 @@ from gns3server.schemas.compute.virtualbox_nodes import (
VirtualBoxOnCloseAction,
)
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -65,5 +66,5 @@ class VirtualBoxTemplate(VirtualBoxTemplateBase):
vmname: str = Field(..., description="VirtualBox VM name (in VirtualBox itself)")
class VirtualBoxTemplateUpdate(VirtualBoxTemplateBase):
pass
class VirtualBoxTemplateUpdate(PartialUpdateModel, VirtualBoxTemplateBase):
update_excluded_fields = ("template_id",)

View File

@ -26,6 +26,7 @@ from gns3server.schemas.compute.vmware_nodes import (
VMwareOnCloseAction,
)
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -63,5 +64,5 @@ class VMwareTemplate(VMwareTemplateBase):
vmx_path: str = Field(..., description="Path to the vmx file")
class VMwareTemplateUpdate(VMwareTemplateBase):
pass
class VMwareTemplateUpdate(PartialUpdateModel, VMwareTemplateBase):
update_excluded_fields = ("template_id",)

View File

@ -21,6 +21,7 @@ from pydantic import Field
from gns3server.schemas.compute.vpcs_nodes import ConsoleType
from ...update import PartialUpdateModel
from . import Category, TemplateBase
@ -35,5 +36,5 @@ class VPCSTemplate(TemplateBase):
)
class VPCSTemplateUpdate(VPCSTemplate):
pass
class VPCSTemplateUpdate(PartialUpdateModel, VPCSTemplate):
update_excluded_fields = ("template_id",)

View File

@ -20,6 +20,7 @@ from uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr, Field, SecretStr
from ..update import PartialUpdateModel
from .base import DateTimeModelMixin
@ -43,7 +44,7 @@ class UserCreate(UserBase):
password: SecretStr = Field(..., min_length=8, max_length=100)
class UserUpdate(UserBase):
class UserUpdate(PartialUpdateModel, UserBase):
"""
Properties to update a user.
"""

View File

@ -0,0 +1,60 @@
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Any, ClassVar, Tuple
from pydantic import BaseModel, ConfigDict, model_validator
def _remove_defaults(schema: dict) -> None:
for property_schema in schema.get("properties", {}).values():
property_schema.pop("default", None)
class PartialUpdateModel(BaseModel):
"""
Turns a model into a partial update model: every field can be omitted, keeps its
type and constraints, and publishes no default. Dumps only include the fields that
were set. The fields listed in `update_excluded_fields` are dropped. Use it as the
first base of an update schema.
"""
model_config = ConfigDict(json_schema_extra=_remove_defaults)
update_excluded_fields: ClassVar[Tuple[str, ...]] = ()
@model_validator(mode="before")
@classmethod
def _drop_excluded_fields(cls, data):
if isinstance(data, dict):
return {key: value for key, value in data.items() if key not in cls.update_excluded_fields}
return data
def model_dump(self, *, exclude_unset: bool = True, **kwargs: Any) -> dict[str, Any]:
return super().model_dump(exclude_unset=exclude_unset, **kwargs)
def model_dump_json(self, *, exclude_unset: bool = True, **kwargs: Any) -> str:
return super().model_dump_json(exclude_unset=exclude_unset, **kwargs)
@classmethod
def __pydantic_init_subclass__(cls, **kwargs) -> None:
super().__pydantic_init_subclass__(**kwargs)
for name in cls.update_excluded_fields:
cls.model_fields.pop(name, None)
for field in cls.model_fields.values():
field.default = None
field.default_factory = None
cls.model_rebuild(force=True)

View File

@ -117,6 +117,28 @@ class TestControllerProjectRoutes:
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test2"
async def test_update_project_ignores_path_and_project_id(
self, app: FastAPI, client: AsyncClient, controller: Controller
) -> None:
params = {"name": "test", "project_id": "10010203-0405-0607-0809-0a0b0c0d0e0f"}
response = await client.post(app.url_path_for("create_project"), json=params)
assert response.status_code == status.HTTP_201_CREATED
path = response.json()["path"]
params = {
"name": "test2",
"path": os.path.join(os.path.dirname(path), "other"),
"project_id": "20010203-0405-0607-0809-0a0b0c0d0e0f",
}
response = await client.put(
app.url_path_for("update_project", project_id="10010203-0405-0607-0809-0a0b0c0d0e0f"), json=params
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "test2"
assert response.json()["path"] == path
assert response.json()["project_id"] == "10010203-0405-0607-0809-0a0b0c0d0e0f"
async def test_update_project_with_variables(
self, app: FastAPI, client: AsyncClient, controller: Controller
) -> None:

View File

@ -0,0 +1,44 @@
import importlib
import inspect
import pkgutil
import pytest
from pydantic import BaseModel
from gns3server import schemas
from gns3server.schemas.update import PartialUpdateModel
def _discover_update_models():
models = {}
for module_info in pkgutil.walk_packages(schemas.__path__, f"{schemas.__name__}."):
module = importlib.import_module(module_info.name)
for name, cls in vars(module).items():
if inspect.isclass(cls) and issubclass(cls, BaseModel) and name.endswith("Update"):
models[f"{cls.__module__}.{cls.__qualname__}"] = cls
return models
UPDATE_MODELS = _discover_update_models()
PARTIAL_UPDATE_MODELS = {k: v for k, v in UPDATE_MODELS.items() if issubclass(v, PartialUpdateModel)}
@pytest.mark.parametrize("model", UPDATE_MODELS.values(), ids=UPDATE_MODELS.keys())
def test_update_schema_publishes_no_defaults(model):
properties = model.model_json_schema()["properties"]
assert {name: p["default"] for name, p in properties.items() if p.get("default") is not None} == {}
@pytest.mark.parametrize("model", PARTIAL_UPDATE_MODELS.values(), ids=PARTIAL_UPDATE_MODELS.keys())
def test_partial_update_schema_drops_excluded_fields(model):
inherited = {field for base in model.__mro__[1:] if hasattr(base, "model_fields") for field in base.model_fields}
for field in model.update_excluded_fields:
assert field in inherited
assert field not in model.model_fields
assert model.model_validate({field: None}).model_dump(exclude_unset=True) == {}
@pytest.mark.parametrize("model", PARTIAL_UPDATE_MODELS.values(), ids=PARTIAL_UPDATE_MODELS.keys())
def test_partial_update_schema_requires_nothing(model):
assert model().model_dump() == {}
assert model().model_dump_json() == "{}"