Fix pyright errors

plugins/quetz_conda_suggest/quetz_conda_suggest/main.py

@@ -53,6 +53,8 @@ def post_add_package_version(version, condainfo):
             db.add(metadata)
         else:
             metadata = db.get(db_models.CondaSuggestMetadata, version.id)
+            if not metadata:
+                raise KeyError(f"Metadata with version '{version.id}' not found")
             metadata.data = json.dumps(suggest_map)
         db.commit()
         generate_channel_suggest_map(db, version.channel_name, subdir)

plugins/quetz_content_trust/quetz_content_trust/api.py

@@ -171,6 +171,9 @@ def get_self_delegation(nullable: bool = False):
             # mamba API when loading the root role from file
             self_delegation = [r for r in db_role.delegations if r.type == "root"][0]
 
+        if not self_delegation:
+            raise RuntimeError("self_delegation must not be None")
+
         # db_role.delegation = self_delegation
         self_delegation.consumers.append(db_role)
 

plugins/quetz_runexports/quetz_runexports/main.py

@@ -24,5 +24,7 @@ def post_add_package_version(version, condainfo):
             db.add(metadata)
         else:
             metadata = db.get(db_models.PackageVersionMetadata, version.id)
+            if not metadata:
+                raise KeyError(f"No metadata found for version '{version.id}'.")
             metadata.data = run_exports
         db.commit()

quetz/dao.py

@@ -11,9 +11,9 @@
 
 from sqlalchemy import and_, func, insert, or_
 from sqlalchemy.dialects.postgresql import insert as pg_insert
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy.exc import IntegrityError, NoResultFound
 from sqlalchemy.ext.compiler import compiles
-from sqlalchemy.orm import Query, Session, aliased, exc, joinedload
+from sqlalchemy.orm import Query, Session, aliased, joinedload
 from sqlalchemy.sql.expression import FunctionElement, Insert
 from sqlalchemy.types import DateTime
 
@@ -193,13 +193,13 @@ def rollback(self):
     def get_profile(self, user_id):
         try:
             return self.db.query(Profile).filter(Profile.user_id == user_id).one()
-        except exc.NoResultFound:
+        except NoResultFound:
             logger.error("User not found")
 
     def get_user(self, user_id):
         try:
             return self.db.query(User).filter(User.id == user_id).one()
-        except exc.NoResultFound:
+        except NoResultFound:
             logger.error("User not found")
 
     def get_users(self, skip: int, limit: int, q: str, order_by: str = 'username:asc'):
@@ -621,6 +621,8 @@ def create_package(
 
     def update_package_channeldata(self, channel_name, package_name, channeldata):
         package = self.get_package(channel_name, package_name)
+        if package is None:
+            raise KeyError(f"Package '{package_name}' not found.")
         if package.channeldata:
             old_data = json.loads(package.channeldata)
         else:
@@ -651,6 +653,8 @@ def get_channel_member(self, channel_name, username):
 
     def create_channel_member(self, channel_name, new_member):
         user = self.get_user_by_username(new_member.username)
+        if user is None:
+            raise KeyError(f"User '{new_member.username}' not found.")
 
         member = ChannelMember(
             channel_name=channel_name, user_id=user.id, role=new_member.role
@@ -681,6 +685,8 @@ def get_package_member(self, channel_name, package_name, username):
 
     def create_package_member(self, channel_name, package_name, new_member):
         user = self.get_user_by_username(new_member.username)
+        if user is None:
+            raise KeyError(f"User '{new_member.username}' not found.")
 
         member = PackageMember(
             channel_name=channel_name,

quetz/main.py

@@ -250,6 +250,8 @@ def dummy_login(
     username: str, dao: Dao = Depends(get_dao), session=Depends(get_session)
 ):
     user = dao.get_user_by_username(username)
+    if user is None:
+        raise HTTPException(status_code=404, detail=f"User '{username}' not found.")
 
     logout(session)
     session["user_id"] = str(uuid.UUID(bytes=user.id))
@@ -439,6 +441,11 @@ def delete_user(
 ):
     user = dao.get_user_by_username(username)
 
+    if not user or not user.profile:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"User {username} not found"
+        )
+
     auth.assert_delete_user(user.id)
     dao.delete_user(user.id)
 
@@ -1106,6 +1113,11 @@ def delete_package_version(
         channel_name, package_name, filename, platform
     )
 
+    if not version or not version.package:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Package version not found"
+        )
+
     auth.assert_package_delete(version.package)
 
     if not version:

quetz/tasks/mirror.py

@@ -459,6 +459,10 @@ def create_packages_from_channeldata(
         except DBError:
             # package already exists so skip it so we retrieve and update it
             package = dao.get_package(channel_name, package_name)
+            if not package:
+                raise KeyError(
+                    f"Package '{package_name}' not found in channel {channel_name}"
+                )
             package.description = description
             package.summary = summary
         package.url = metadata.get("home", "")

quetz/tasks/workers.py

@@ -154,6 +154,8 @@ def job_wrapper(
     user_id: Optional[str]
     if task_id:
         task = db.query(Task).filter(Task.id == task_id).one_or_none()
+        if not task:
+            raise KeyError(f"Task '{task_id}' not found")
         # take extra arguments from job definition
         if task.job.extra_args:
             job_extra_args = json.loads(task.job.extra_args)