fix(typing): resolve mypy errors in db.repositories.images

gns3server/db/models/images.py

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from sqlalchemy import Table, Column, String, ForeignKey, BigInteger, Integer
-from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base, BaseTable, GUID
 
@@ -34,7 +34,7 @@ class Image(BaseTable):
 
     image_id = Column(Integer, primary_key=True, autoincrement=True)
     filename = Column(String, index=True)
-    path = Column(String, unique=True)
+    path: Mapped[str] = mapped_column(String, unique=True, nullable=True)
     image_type = Column(String)
     image_size = Column(BigInteger)
     checksum = Column(String, index=True)

gns3server/db/repositories/images.py

@@ -17,8 +17,9 @@
 
 import os
 
-from typing import Optional, List
+from typing import Optional, List, cast
 from sqlalchemy import select, delete, update
+from sqlalchemy.engine import CursorResult
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from .base import BaseRepository
@@ -50,7 +51,7 @@ class ImagesRepository(BaseRepository):
         result = await self._db_session.execute(query)
         return result.scalars().one_or_none()
 
-    async def get_image_by_checksum(self, checksum: str, image_dir: str = None) -> Optional[models.Image]:
+    async def get_image_by_checksum(self, checksum: str, image_dir: Optional[str] = None) -> Optional[models.Image]:
         """
         Get an image by its checksum.
         """
@@ -76,7 +77,7 @@ class ImagesRepository(BaseRepository):
         else:
             query = select(models.Image)
         result = await self._db_session.execute(query)
-        return result.scalars().all()
+        return list(result.scalars().all())
 
     async def get_image_templates(self, image_id: int) -> Optional[List[models.Template]]:
         """
@@ -86,7 +87,7 @@ class ImagesRepository(BaseRepository):
         query = select(models.Template).join(models.Template.images).filter(models.Image.image_id == image_id)
 
         result = await self._db_session.execute(query)
-        return result.scalars().all()
+        return list(result.scalars().all())
 
     async def add_image(self, image_name, image_type, image_size, path, checksum, checksum_algorithm) -> models.Image:
         """
@@ -108,7 +109,7 @@ class ImagesRepository(BaseRepository):
         await self._db_session.refresh(db_image)
         return db_image
 
-    async def update_image(self, image_path: str, checksum: str, checksum_algorithm: str) -> models.Image:
+    async def update_image(self, image_path: str, checksum: str, checksum_algorithm: str) -> Optional[models.Image]:
         """
         Update an image.
         """
@@ -142,9 +143,9 @@ class ImagesRepository(BaseRepository):
             query = delete(models.Image).where(models.Image.filename == image_name)
         result = await self._db_session.execute(query)
         await self._db_session.commit()
-        return result.rowcount > 0
+        return cast(CursorResult, result).rowcount > 0
 
-    async def prune_images(self, skip_images: list[str] = None) -> int:
+    async def prune_images(self, skip_images: Optional[list[str]] = None) -> int:
         """
         Prune images not attached to any template.
         """