diff --git a/gns3server/db/models/templates.py b/gns3server/db/models/templates.py
index 62ea74a9e..0450be881 100644
--- a/gns3server/db/models/templates.py
+++ b/gns3server/db/models/templates.py
@@ -16,19 +16,26 @@
# along with this program. If not, see .
+import uuid
+from typing import Optional
+
from sqlalchemy import Boolean, Column, String, Integer, Float, ForeignKey, JSON
-from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base import BaseTable, generate_uuid, GUID
from .images import image_template_map
+def template_id_column() -> Mapped[uuid.UUID]:
+ return mapped_column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+
+
class Template(BaseTable):
__tablename__ = "templates"
- template_id = Column(GUID, primary_key=True, default=generate_uuid)
+ template_id: Mapped[uuid.UUID] = mapped_column(GUID, primary_key=True, default=generate_uuid)
name = Column(String, index=True)
- version = Column(String)
+ version: Mapped[Optional[str]] = mapped_column(String)
category = Column(String)
default_name_format = Column(String)
symbol = Column(String)
@@ -36,7 +43,7 @@ class Template(BaseTable):
usage = Column(String)
netmiko_device_type = Column(String)
appliance_metadata = Column(JSON)
- template_type = Column(String)
+ template_type: Mapped[str] = mapped_column(String, nullable=True)
tags = Column(JSON)
compute_id = Column(String)
images = relationship("Image", secondary=image_template_map, back_populates="templates")
@@ -50,7 +57,7 @@ class Template(BaseTable):
class CloudTemplate(Template):
__tablename__ = "cloud_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
ports_mapping = Column(JSON)
remote_console_host = Column(String)
remote_console_port = Column(Integer)
@@ -63,7 +70,7 @@ class CloudTemplate(Template):
class DockerTemplate(Template):
__tablename__ = "docker_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
image = Column(String)
adapters = Column(Integer)
mac_address = Column(String)
@@ -88,10 +95,10 @@ class DockerTemplate(Template):
class DynamipsTemplate(Template):
__tablename__ = "dynamips_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
- platform = Column(String)
+ template_id: Mapped[uuid.UUID] = template_id_column()
+ platform: Mapped[str] = mapped_column(String, nullable=True)
chassis = Column(String)
- image = Column(String)
+ image: Mapped[str] = mapped_column(String, nullable=True)
exec_area = Column(Integer)
mmap = Column(Boolean)
mac_addr = Column(String)
@@ -130,7 +137,7 @@ class DynamipsTemplate(Template):
class EthernetHubTemplate(Template):
__tablename__ = "ethernet_hub_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
ports_mapping = Column(JSON)
__mapper_args__ = {"polymorphic_identity": "ethernet_hub", "polymorphic_load": "selectin"}
@@ -139,7 +146,7 @@ class EthernetHubTemplate(Template):
class EthernetSwitchTemplate(Template):
__tablename__ = "ethernet_switch_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
ports_mapping = Column(JSON)
console_type = Column(String)
@@ -149,8 +156,8 @@ class EthernetSwitchTemplate(Template):
class IOUTemplate(Template):
__tablename__ = "iou_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
- path = Column(String)
+ template_id: Mapped[uuid.UUID] = template_id_column()
+ path: Mapped[str] = mapped_column(String, nullable=True)
ethernet_adapters = Column(Integer)
serial_adapters = Column(Integer)
ram = Column(Integer)
@@ -168,7 +175,7 @@ class IOUTemplate(Template):
class QemuTemplate(Template):
__tablename__ = "qemu_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
qemu_path = Column(String)
platform = Column(String)
linked_clone = Column(Boolean)
@@ -214,7 +221,7 @@ class QemuTemplate(Template):
class VirtualBoxTemplate(Template):
__tablename__ = "virtualbox_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
vmname = Column(String)
ram = Column(Integer)
linked_clone = Column(Boolean)
@@ -236,7 +243,7 @@ class VirtualBoxTemplate(Template):
class VMwareTemplate(Template):
__tablename__ = "vmware_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
vmx_path = Column(String)
linked_clone = Column(Boolean)
first_port_name = Column(String)
@@ -257,7 +264,7 @@ class VMwareTemplate(Template):
class VPCSTemplate(Template):
__tablename__ = "vpcs_templates"
- template_id = Column(GUID, ForeignKey("templates.template_id", ondelete="CASCADE"), primary_key=True)
+ template_id: Mapped[uuid.UUID] = template_id_column()
base_script_file = Column(String)
console_type = Column(String)
console_auto_start = Column(Boolean, default=False)
diff --git a/gns3server/db/repositories/templates.py b/gns3server/db/repositories/templates.py
index 582404301..9d5c1a9af 100644
--- a/gns3server/db/repositories/templates.py
+++ b/gns3server/db/repositories/templates.py
@@ -16,11 +16,13 @@
# along with this program. If not, see .
import os
+import uuid
import logging
from uuid import UUID
-from typing import List, Union, Optional
+from typing import List, Union, Optional, cast
from sqlalchemy import select, delete
+from sqlalchemy.engine import CursorResult
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from sqlalchemy.orm.session import make_transient
@@ -64,7 +66,7 @@ class TemplatesRepository(BaseRepository):
result = await self._db_session.execute(query)
return result.scalars().first()
- async def get_template_by_name_and_version(self, name: str, version: str) -> Union[None, models.Template]:
+ async def get_template_by_name_and_version(self, name: str, version: Optional[str]) -> Union[None, models.Template]:
query = (
select(models.Template)
@@ -89,7 +91,7 @@ class TemplatesRepository(BaseRepository):
query = select(models.Template).options(selectinload(models.Template.images))
result = await self._db_session.execute(query)
- return result.scalars().all()
+ return list(result.scalars().all())
async def create_template(self, template_type: str, template_settings: dict) -> models.Template:
@@ -100,7 +102,7 @@ class TemplatesRepository(BaseRepository):
await self._db_session.refresh(db_template)
return db_template
- async def update_template(self, db_template: models.Template, template_settings: dict) -> schemas.Template:
+ async def update_template(self, db_template: models.Template, template_settings: dict) -> models.Template:
# update the fields directly because update() query couldn't work
for key, value in template_settings.items():
@@ -114,9 +116,9 @@ class TemplatesRepository(BaseRepository):
query = delete(models.Template).where(models.Template.template_id == template_id)
result = await self._db_session.execute(query)
await self._db_session.commit()
- return result.rowcount > 0
+ return cast(CursorResult, result).rowcount > 0
- async def duplicate_template(self, template_id: UUID) -> Optional[schemas.Template]:
+ async def duplicate_template(self, template_id: UUID) -> Optional[models.Template]:
query = (
select(models.Template)
@@ -128,7 +130,7 @@ class TemplatesRepository(BaseRepository):
# duplicate db object with new primary key (template_id)
self._db_session.expunge(db_template)
make_transient(db_template)
- db_template.template_id = None
+ db_template.template_id = uuid.uuid4()
self._db_session.add(db_template)
await self._db_session.commit()
await self._db_session.refresh(db_template)
@@ -205,4 +207,4 @@ class TemplatesRepository(BaseRepository):
query = select(models.Image).join(models.Image.templates).filter(models.Template.template_id == template_id)
result = await self._db_session.execute(query)
- return result.scalars().all()
+ return list(result.scalars().all())
diff --git a/gns3server/services/templates.py b/gns3server/services/templates.py
index 10a167c89..00f02b4b0 100644
--- a/gns3server/services/templates.py
+++ b/gns3server/services/templates.py
@@ -20,7 +20,7 @@ import pydantic
from uuid import UUID
from fastapi.encoders import jsonable_encoder
-from typing import List
+from typing import List, Optional
from gns3server import schemas
from gns3server.config import Config
@@ -34,7 +34,7 @@ from gns3server.controller.controller_error import (
)
-TEMPLATE_TYPE_TO_SCHEMA = {
+TEMPLATE_TYPE_TO_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
"cloud": schemas.CloudTemplate,
"ethernet_hub": schemas.EthernetHubTemplate,
"ethernet_switch": schemas.EthernetSwitchTemplate,
@@ -47,7 +47,7 @@ TEMPLATE_TYPE_TO_SCHEMA = {
"qemu": schemas.QemuTemplate,
}
-TEMPLATE_TYPE_TO_UPDATE_SCHEMA = {
+TEMPLATE_TYPE_TO_UPDATE_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
"cloud": schemas.CloudTemplateUpdate,
"ethernet_hub": schemas.EthernetHubTemplateUpdate,
"ethernet_switch": schemas.EthernetSwitchTemplateUpdate,
@@ -59,7 +59,7 @@ TEMPLATE_TYPE_TO_UPDATE_SCHEMA = {
"qemu": schemas.QemuTemplateUpdate,
}
-DYNAMIPS_PLATFORM_TO_SCHEMA = {
+DYNAMIPS_PLATFORM_TO_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
"c7200": schemas.C7200DynamipsTemplate,
"c3745": schemas.C3745DynamipsTemplate,
"c3725": schemas.C3725DynamipsTemplate,
@@ -69,7 +69,7 @@ DYNAMIPS_PLATFORM_TO_SCHEMA = {
"c1700": schemas.C1700DynamipsTemplate,
}
-DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA = {
+DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA: dict[str, type[pydantic.BaseModel]] = {
"c7200": schemas.C7200DynamipsTemplateUpdate,
"c3745": schemas.C3745DynamipsTemplateUpdate,
"c3725": schemas.C3725DynamipsTemplateUpdate,
@@ -169,11 +169,12 @@ class TemplatesService:
for builtin_template in BUILTIN_TEMPLATES:
builtin_template["symbol"] = self._controller.symbols.resolve_symbol(builtin_template["symbol"])
- def get_builtin_template(self, template_id: UUID) -> dict:
+ def get_builtin_template(self, template_id: UUID) -> Optional[dict]:
for builtin_template in BUILTIN_TEMPLATES:
if builtin_template["template_id"] == template_id:
return jsonable_encoder(builtin_template)
+ return None
def _base_path(self):
return self._templates_repo.configs_path()
@@ -303,7 +304,7 @@ class TemplatesService:
try:
# validate the update settings
update_settings = jsonable_encoder(template_update, exclude_unset=True)
- if db_template.template_type == "dynamips":
+ if isinstance(db_template, models.DynamipsTemplate):
template_schema = DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA[db_template.platform]
else:
template_schema = TEMPLATE_TYPE_TO_UPDATE_SCHEMA[db_template.template_type]
@@ -312,9 +313,9 @@ class TemplatesService:
raise ControllerBadRequestError(f"JSON schema error received while updating template: {e}")
images_to_add_to_template = await self._find_images(db_template.template_type, template_settings)
- if db_template.template_type == "dynamips" and "image" in template_settings:
+ if isinstance(db_template, models.DynamipsTemplate) and "image" in template_settings:
await self._remove_image(db_template.template_id, db_template.image)
- elif db_template.template_type == "iou" and "path" in template_settings:
+ elif isinstance(db_template, models.IOUTemplate) and "path" in template_settings:
await self._remove_image(db_template.template_id, db_template.path)
elif db_template.template_type == "qemu":
for key in template_update.model_dump().keys():