Merge pull request #2930 from markparonyan/mypy-db-templates-images

fix(typing): resolve the remaining mypy errors (18 modules)

gns3server/api/routes/controller/images.py

@@ -38,6 +38,7 @@ from gns3server.utils.images import (
     default_images_directory,
     get_builtin_disks,
 )
+import gns3server.db.models as models
 from gns3server.db.repositories.images import ImagesRepository
 from gns3server.db.repositories.templates import TemplatesRepository
 from gns3server.db.repositories.rbac import RbacRepository
@@ -68,7 +69,7 @@ async def create_qemu_image(
     image_path: str,
     image_data: schemas.QemuDiskImageCreate,
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
-) -> schemas.Image:
+) -> models.Image:
     """
     Create a new blank Qemu image.
 
@@ -115,7 +116,7 @@ async def create_qemu_image(
 async def get_images(
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
     image_type: Optional[schemas.ImageType] = None,
-) -> List[schemas.Image]:
+) -> List[models.Image]:
     """
     Return all images.
 
@@ -139,7 +140,7 @@ async def upload_image(
     current_user: schemas.User = Depends(get_current_active_user),
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
     install_appliances: Optional[bool] = False,
-) -> schemas.Image:
+) -> models.Image:
     """
     Upload an image.
 
@@ -256,7 +257,7 @@ async def install_images(
 async def get_image(
     image_path: str,
     images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
-) -> schemas.Image:
+) -> models.Image:
     """
     Return an image.
 
@@ -299,7 +300,7 @@ async def delete_image(
 
     templates = await images_repo.get_image_templates(image.image_id)
     if templates:
-        template_names = ", ".join([template.name for template in templates])
+        template_names = ", ".join([str(template.name) for template in templates])
         raise ControllerError(f"Image '{image_path}' is used by one or more templates: {template_names}")
 
     project_names = Controller.instance().find_projects_using_image(image.filename)

gns3server/api/routes/controller/llm_model_configs.py

@@ -206,7 +206,11 @@ async def create_user_llm_model_config(
         # Extract config fields (excluding table-level fields)
         config_fields = config_create.model_dump(exclude={"name", "model_type", "is_default"})
         new_config = await llm_repo.create_user_config(
-            user_id, config_create.name, config_create.model_type, config_fields, is_default=config_create.is_default
+            user_id,
+            config_create.name,
+            config_create.model_type,
+            config_fields,
+            is_default=bool(config_create.is_default),
         )
 
         return schemas.LLMModelConfigResponse(
@@ -346,6 +350,10 @@ async def set_user_default_llm_model_config(
 
         # Get the updated config
         config = await llm_repo.get_user_config(config_id)
+        if config is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail=f"LLM model configuration '{config_id}' not found"
+            )
         return schemas.LLMModelConfigResponse(
             config_id=config.config_id,
             name=config.name,
@@ -506,7 +514,11 @@ async def create_group_llm_model_config(
         # Extract config fields (excluding table-level fields)
         config_fields = config_create.model_dump(exclude={"name", "model_type", "is_default"})
         new_config = await llm_repo.create_group_config(
-            group_id, config_create.name, config_create.model_type, config_fields, is_default=config_create.is_default
+            group_id,
+            config_create.name,
+            config_create.model_type,
+            config_fields,
+            is_default=bool(config_create.is_default),
         )
 
         return schemas.LLMModelConfigResponse(
@@ -646,6 +658,10 @@ async def set_group_default_llm_model_config(
 
         # Get the updated config
         config = await llm_repo.get_group_config(config_id)
+        if config is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail=f"LLM model configuration '{config_id}' not found"
+            )
         return schemas.LLMModelConfigResponse(
             config_id=config.config_id,
             name=config.name,

gns3server/api/routes/controller/pools.py

@@ -21,12 +21,13 @@ API routes for resource pools.
 
 from fastapi import APIRouter, Depends, status
 from uuid import UUID
-from typing import List
+from typing import List, Optional
 
 from gns3server import schemas
 from gns3server.controller.controller_error import ControllerError, ControllerBadRequestError, ControllerNotFoundError
 
 from gns3server.controller import Controller
+import gns3server.db.models as models
 from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.db.repositories.pools import ResourcePoolsRepository
 
@@ -43,7 +44,7 @@ router = APIRouter()
 @router.get("", response_model=List[schemas.ResourcePool], dependencies=[Depends(has_privilege("Pool.Audit"))])
 async def get_resource_pools(
     pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
-) -> List[schemas.ResourcePool]:
+) -> List[models.ResourcePool]:
     """
     Get all resource pools.
 
@@ -62,7 +63,7 @@ async def get_resource_pools(
 async def create_resource_pool(
     resource_pool_create: schemas.ResourcePoolCreate,
     pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
-) -> schemas.ResourcePool:
+) -> models.ResourcePool:
     """
     Create a new resource pool
 
@@ -80,7 +81,7 @@ async def create_resource_pool(
 )
 async def get_resource_pool(
     resource_pool_id: UUID, pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository))
-) -> schemas.ResourcePool:
+) -> models.ResourcePool:
     """
     Get a resource pool.
 
@@ -100,7 +101,7 @@ async def update_resource_pool(
     resource_pool_id: UUID,
     resource_pool_update: schemas.ResourcePoolUpdate,
     pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
-) -> schemas.ResourcePool:
+) -> Optional[models.ResourcePool]:
     """
     Update a resource pool.
 
@@ -176,7 +177,7 @@ async def delete_resource_pool(
 async def get_pool_resources(
     resource_pool_id: UUID,
     pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
-) -> List[schemas.Resource]:
+) -> List[models.Resource]:
     """
     Get all resource in a pool.
 
@@ -215,13 +216,13 @@ async def add_resource_to_pool(
     # we only support projects in resource pools for now
     project = Controller.instance().get_project(str(resource_id))
 
-    resource = await pools_repo.get_resource(resource_id)
-    if not resource:
+    db_resource = await pools_repo.get_resource(resource_id)
+    if not db_resource:
         # the resource is not in the database yet, create it
         resource_create = schemas.ResourceCreate(resource_id=resource_id, resource_type="project", name=project.name)
-        resource = await pools_repo.create_resource(resource_create)
+        db_resource = await pools_repo.create_resource(resource_create)
 
-    await pools_repo.add_resource_to_pool(resource_pool_id, resource)
+    await pools_repo.add_resource_to_pool(resource_pool_id, db_resource)
 
 
 @router.delete(

gns3server/api/routes/controller/projects.py

@@ -34,7 +34,7 @@ from fastapi import APIRouter, Depends, Request, Body, HTTPException, status, We
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse, FileResponse
 from websockets.exceptions import ConnectionClosed, WebSocketException
-from typing import List, Optional
+from typing import Any, List, Optional
 from uuid import UUID
 
 from gns3server import schemas
@@ -57,7 +57,9 @@ from .dependencies.rbac import has_privilege, has_privilege_on_websocket
 from .dependencies.authentication import get_current_active_user
 from .dependencies.database import get_repository
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -636,8 +638,8 @@ async def export_project(
     include_images: bool = False,
     reset_mac_addresses: bool = False,
     keep_compute_ids: bool = False,
-    compression: schemas.ProjectCompression = "zstd",
-    compression_level: int = None,
+    compression: schemas.ProjectCompression = schemas.ProjectCompression.zstd,
+    compression_level: Optional[int] = None,
 ) -> StreamingResponse:
     """
     Export a project as a portable archive.
@@ -650,19 +652,19 @@ async def export_project(
 
     compression_query = compression.lower()
     if compression_query == "zip":
-        compression = zipfile.ZIP_DEFLATED
+        zip_compression = zipfile.ZIP_DEFLATED
         if compression_level is not None and (compression_level < 0 or compression_level > 9):
             raise ControllerBadRequestError("Compression level must be between 0 and 9 for ZIP compression")
     elif compression_query == "none":
-        compression = zipfile.ZIP_STORED
+        zip_compression = zipfile.ZIP_STORED
     elif compression_query == "bzip2":
-        compression = zipfile.ZIP_BZIP2
+        zip_compression = zipfile.ZIP_BZIP2
         if compression_level is not None and (compression_level < 1 or compression_level > 9):
             raise ControllerBadRequestError("Compression level must be between 1 and 9 for BZIP2 compression")
     elif compression_query == "lzma":
-        compression = zipfile.ZIP_LZMA
+        zip_compression = zipfile.ZIP_LZMA
     elif compression_query == "zstd":
-        compression = zipfile.ZIP_ZSTANDARD
+        zip_compression = zipfile.ZIP_ZSTANDARD
         if compression_level is not None and (compression_level < 1 or compression_level > 22):
             raise ControllerBadRequestError("Compression level must be between 1 and 22 for Zstandard compression")
 
@@ -681,7 +683,7 @@ async def export_project(
                 f"Exporting project '{project.name}' with '{compression_query}' compression (level {compression_level})"
             )
             with tempfile.TemporaryDirectory(dir=working_dir) as tmpdir:
-                with aiozipstream.ZipFile(compression=compression, compresslevel=compression_level) as zstream:
+                with aiozipstream.ZipFile(compression=zip_compression, compresslevel=compression_level) as zstream:
                     await export_controller_project(
                         zstream,
                         project,

gns3server/api/routes/controller/templates.py

@@ -27,7 +27,7 @@ import logging
 log = logging.getLogger(__name__)
 
 from fastapi import APIRouter, Request, HTTPException, Depends, Response, status, Query
-from typing import List, Optional
+from typing import Any, List, Optional, Union
 from uuid import UUID
 
 from gns3server import schemas
@@ -43,7 +43,9 @@ from .dependencies.authentication import get_current_active_user
 from .dependencies.rbac import has_privilege
 from .dependencies.database import get_repository
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find template"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find template"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -57,7 +59,7 @@ router = APIRouter(responses=responses)
 async def create_template(
     template_create: schemas.TemplateCreate,
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
-) -> schemas.Template:
+) -> dict:
     """
     Create a new template.
 
@@ -80,7 +82,7 @@ async def get_template(
     request: Request,
     response: Response,
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
-) -> schemas.Template:
+) -> Union[dict, Response]:
     """
     Return a template.
 
@@ -108,7 +110,7 @@ async def update_template(
     template_id: UUID,
     template_update: schemas.TemplateUpdate,
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
-) -> schemas.Template:
+) -> dict:
     """
     Update a template.
 
@@ -125,7 +127,7 @@ async def delete_template(
     template_id: UUID,
     prune_images: Optional[bool] = False,
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
-    images_repo: RbacRepository = Depends(get_repository(ImagesRepository)),
+    images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
     rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
 ) -> None:
     """
@@ -155,7 +157,7 @@ async def delete_template(
                 if str(template.template_id) != str(template_id)
             ]
             if other_templates:
-                template_names = ", ".join([template.name for template in other_templates])
+                template_names = ", ".join([str(template.name) for template in other_templates])
                 raise ControllerError(f"Image '{image.path}' is used by one or more templates: {template_names}")
 
             if referenced_filenames is None:
@@ -193,7 +195,7 @@ async def get_templates(
     templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
     current_user: schemas.User = Depends(get_current_active_user),
     tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
-) -> List[schemas.Template]:
+) -> List[dict]:
     """
     Return all templates.
 
@@ -244,7 +246,7 @@ async def get_templates(
 )
 async def duplicate_template(
     template_id: UUID, templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository))
-) -> schemas.Template:
+) -> dict:
     """
     Duplicate a template.
 

gns3server/api/routes/controller/users.py

@@ -22,7 +22,7 @@ API routes for users.
 from fastapi import APIRouter, Depends, HTTPException, Response, status
 from fastapi.security import OAuth2PasswordRequestForm
 from uuid import UUID
-from typing import List
+from typing import List, Optional
 
 from gns3server import schemas
 from gns3server.controller.controller_error import (
@@ -32,6 +32,7 @@ from gns3server.controller.controller_error import (
     ControllerForbiddenError,
 )
 
+import gns3server.db.models as models
 from gns3server.db.repositories.users import UsersRepository
 from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.services import auth_service
@@ -168,7 +169,7 @@ async def update_logged_in_user(
     user_update: schemas.LoggedInUserUpdate,
     current_user: schemas.User = Depends(get_current_active_user),
     users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
-) -> schemas.User:
+) -> Optional[models.User]:
     """
     Update the current active user.
     """
@@ -180,7 +181,7 @@ async def update_logged_in_user(
 
 
 @router.get("", response_model=List[schemas.User], dependencies=[Depends(has_privilege("User.Audit"))])
-async def get_users(users_repo: UsersRepository = Depends(get_repository(UsersRepository))) -> List[schemas.User]:
+async def get_users(users_repo: UsersRepository = Depends(get_repository(UsersRepository))) -> List[models.User]:
     """
     Get all users.
 
@@ -198,7 +199,7 @@ async def get_users(users_repo: UsersRepository = Depends(get_repository(UsersRe
 )
 async def create_user(
     user_create: schemas.UserCreate, users_repo: UsersRepository = Depends(get_repository(UsersRepository))
-) -> schemas.User:
+) -> models.User:
     """
     Create a new user.
 
@@ -218,7 +219,7 @@ async def create_user(
 async def get_user(
     user_id: UUID,
     users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
-) -> schemas.User:
+) -> models.User:
     """
     Get a user.
 
@@ -236,7 +237,7 @@ async def update_user(
     user_id: UUID,
     user_update: schemas.UserUpdate,
     users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
-) -> schemas.User:
+) -> Optional[models.User]:
     """
     Update a user.
 
@@ -287,7 +288,7 @@ async def delete_user(
 )
 async def get_user_memberships(
     user_id: UUID, users_repo: UsersRepository = Depends(get_repository(UsersRepository))
-) -> List[schemas.UserGroup]:
+) -> List[models.UserGroup]:
     """
     Get user memberships.