Merge pull request #2927 from markparonyan/mypy-schemas-plus

fix(typing): resolve mypy errors in schemas + schemas.controller + services + utils.asyncio

gns3server/db/repositories/computes.py

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from uuid import UUID
-from typing import Optional, List
+from typing import Optional, List, Union
 from sqlalchemy import select, update, delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -31,7 +31,7 @@ class ComputesRepository(BaseRepository):
 
         super().__init__(db_session)
 
-    async def get_compute(self, compute_id: UUID) -> Optional[models.Compute]:
+    async def get_compute(self, compute_id: Union[str, UUID]) -> Optional[models.Compute]:
 
         query = select(models.Compute).where(models.Compute.compute_id == compute_id)
         result = await self._db_session.execute(query)
@@ -65,7 +65,9 @@ class ComputesRepository(BaseRepository):
         await self._db_session.refresh(db_compute)
         return db_compute
 
-    async def update_compute(self, compute_id: UUID, compute_update: schemas.ComputeUpdate) -> Optional[models.Compute]:
+    async def update_compute(
+        self, compute_id: Union[str, UUID], compute_update: schemas.ComputeUpdate
+    ) -> Optional[models.Compute]:
 
         update_values = compute_update.model_dump(exclude_unset=True)
         if compute_update.password is not None:
@@ -80,7 +82,7 @@ class ComputesRepository(BaseRepository):
             await self._db_session.refresh(compute_db)  # force refresh of updated_at value
         return compute_db
 
-    async def delete_compute(self, compute_id: UUID) -> bool:
+    async def delete_compute(self, compute_id: Union[str, UUID]) -> bool:
 
         query = delete(models.Compute).where(models.Compute.compute_id == compute_id)
         result = await self._db_session.execute(query)

gns3server/schemas/controller/templates/iou_templates.py

@@ -22,11 +22,11 @@ from pydantic import Field
 from typing import Optional
 
 
-class IOUTemplate(TemplateBase):
+class IOUTemplateBase(TemplateBase):
     category: Optional[Category] = Category.router
     default_name_format: Optional[str] = "IOU{0}"
     symbol: Optional[str] = "multilayer_switch"
-    path: str = Field(..., description="Path of IOU executable")
+    path: Optional[str] = Field(None, description="Path of IOU executable")
     ethernet_adapters: Optional[int] = Field(2, ge=0, description="Number of ethernet adapters")
     serial_adapters: Optional[int] = Field(2, ge=0, description="Number of serial adapters")
     ram: Optional[int] = Field(1024, gt=0, description="Amount of RAM in MB")
@@ -44,5 +44,9 @@ class IOUTemplate(TemplateBase):
     )
 
 
-class IOUTemplateUpdate(IOUTemplate):
-    path: Optional[str] = Field(None, description="Path of IOU executable")
+class IOUTemplate(IOUTemplateBase):
+    path: str = Field(..., description="Path of IOU executable")
+
+
+class IOUTemplateUpdate(IOUTemplateBase):
+    pass

gns3server/schemas/controller/templates/virtualbox_templates.py

@@ -26,11 +26,11 @@ from pydantic import Field
 from typing import Optional, List
 
 
-class VirtualBoxTemplate(TemplateBase):
+class VirtualBoxTemplateBase(TemplateBase):
     category: Optional[Category] = Category.guest
     default_name_format: Optional[str] = "{name}-{0}"
     symbol: Optional[str] = "vbox_guest"
-    vmname: str = Field(..., description="VirtualBox VM name (in VirtualBox itself)")
+    vmname: Optional[str] = Field(None, description="VirtualBox VM name (in VirtualBox itself)")
     ram: Optional[int] = Field(256, gt=0, description="Amount of RAM in MB")
     linked_clone: Optional[bool] = Field(False, description="Whether the VM is a linked clone or not")
     adapters: Optional[int] = Field(
@@ -59,5 +59,9 @@ class VirtualBoxTemplate(TemplateBase):
     custom_adapters: Optional[List[CustomAdapter]] = Field(default_factory=list, description="Custom adapters")
 
 
-class VirtualBoxTemplateUpdate(VirtualBoxTemplate):
-    vmname: Optional[str] = Field(None, description="VirtualBox VM name (in VirtualBox itself)")
+class VirtualBoxTemplate(VirtualBoxTemplateBase):
+    vmname: str = Field(..., description="VirtualBox VM name (in VirtualBox itself)")
+
+
+class VirtualBoxTemplateUpdate(VirtualBoxTemplateBase):
+    pass

gns3server/schemas/controller/templates/vmware_templates.py

@@ -27,11 +27,11 @@ from pydantic import Field
 from typing import Optional, List
 
 
-class VMwareTemplate(TemplateBase):
+class VMwareTemplateBase(TemplateBase):
     category: Optional[Category] = Category.guest
     default_name_format: Optional[str] = "{name}-{0}"
     symbol: Optional[str] = "vmware_guest"
-    vmx_path: str = Field(..., description="Path to the vmx file")
+    vmx_path: Optional[str] = Field(None, description="Path to the vmx file")
     linked_clone: Optional[bool] = Field(False, description="Whether the VM is a linked clone or not")
     first_port_name: Optional[str] = Field("", description="Optional name of the first networking port example: eth0")
     port_name_format: Optional[str] = Field(
@@ -57,5 +57,9 @@ class VMwareTemplate(TemplateBase):
     custom_adapters: Optional[List[CustomAdapter]] = Field(default_factory=list, description="Custom adapters")
 
 
-class VMwareTemplateUpdate(VMwareTemplate):
-    vmx_path: Optional[str] = Field(None, description="Path to the vmx file")
+class VMwareTemplate(VMwareTemplateBase):
+    vmx_path: str = Field(..., description="Path to the vmx file")
+
+
+class VMwareTemplateUpdate(VMwareTemplateBase):
+    pass

gns3server/schemas/qemu_disk_image.py

@@ -78,8 +78,8 @@ class QemuDiskImageAdapterType(str, Enum):
 
 
 class QemuDiskImageBase(BaseModel):
-    format: QemuDiskImageFormat = Field(..., description="Image format type")
-    size: int = Field(..., description="Image size in Megabytes")
+    format: Optional[QemuDiskImageFormat] = Field(None, description="Image format type")
+    size: Optional[int] = Field(None, description="Image size in Megabytes")
     preallocation: Optional[QemuDiskImagePreallocation] = None
     cluster_size: Optional[int] = None
     refcount_bits: Optional[int] = None
@@ -91,10 +91,9 @@ class QemuDiskImageBase(BaseModel):
 
 
 class QemuDiskImageCreate(QemuDiskImageBase):
-    pass
+    format: QemuDiskImageFormat = Field(..., description="Image format type")
+    size: int = Field(..., description="Image size in Megabytes")
 
 
 class QemuDiskImageUpdate(QemuDiskImageBase):
-    format: Optional[QemuDiskImageFormat] = Field(None, description="Image format type")
-    size: Optional[int] = Field(None, description="Image size in Megabytes")
     extend: Optional[int] = Field(None, description="Number of Megabytes to extend the image")

gns3server/utils/asyncio/aiozipstream.py

@@ -32,16 +32,14 @@ import asyncio
 import aiofiles
 from concurrent import futures
 
-from zipfile import (
-    structCentralDir,
-    structEndArchive64,
-    structEndArchive,
-    structEndArchive64Locator,
-    stringCentralDir,
-    stringEndArchive64,
-    stringEndArchive,
-    stringEndArchive64Locator,
-)
+structCentralDir = "<4s4B4HL2L5H2L"
+structEndArchive64 = "<4sQ2H2L4Q"
+structEndArchive = b"<4s4H2LH"
+structEndArchive64Locator = "<4sLQL"
+stringCentralDir = b"PK\x01\x02"
+stringEndArchive64 = b"PK\x06\x06"
+stringEndArchive = b"PK\x05\x06"
+stringEndArchive64Locator = b"PK\x06\x07"
 
 
 ZIP_ZSTANDARD = 93  # zstandard is supported by WinZIP v24 and later, PowerArchiver 2021 and 7-Zip-zstd

pyproject.toml

@@ -313,13 +313,6 @@ module = [
   "gns3server.db.repositories.users",  # 12
   "gns3server.db.tasks",  # 12
   "gns3server.schemas.controller.computes",  # 6
-  "gns3server.schemas.controller.templates.iou_templates",  # 1
-  "gns3server.schemas.controller.templates.virtualbox_templates",  # 1
-  "gns3server.schemas.controller.templates.vmware_templates",  # 1
-  "gns3server.schemas.qemu_disk_image",  # 2
-  "gns3server.services.computes",  # 4
-  "gns3server.services.templates",  # 18
-  "gns3server.utils.asyncio.aiozipstream",  # 8
   "gns3server.utils.http_client",  # 4
   "gns3server.utils.websocket_to_websocket",  # 12
   "gns3server.utils.zipfile_zstd._patcher",  # 1