Fixes for mypy 1.12.0

lib/galaxy/managers/notification.py

@@ -414,16 +414,14 @@ def update_broadcasted_notification(self, notification_id: int, request: Notific
 
     def get_user_notification_preferences(self, user: User) -> UserNotificationPreferences:
         """Gets the user's current notification preferences or the default ones if no preferences are set."""
-        current_notification_preferences = (
-            user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME]
-            if NOTIFICATION_PREFERENCES_SECTION_NAME in user.preferences
-            else None
-        )
-        try:
-            return UserNotificationPreferences.model_validate_json(current_notification_preferences)
-        except ValidationError:
-            # Gracefully return default preferences is they don't exist or get corrupted
-            return UserNotificationPreferences.default()
+        current_notification_preferences = user.preferences.get(NOTIFICATION_PREFERENCES_SECTION_NAME)
+        if current_notification_preferences:
+            try:
+                return UserNotificationPreferences.model_validate_json(current_notification_preferences)
+            except ValidationError:
+                pass
+        # Gracefully return default preferences is they don't exist or get corrupted
+        return UserNotificationPreferences.default()
 
     def update_user_notification_preferences(
         self, user: User, request: UpdateUserNotificationPreferencesRequest

lib/galaxy/tool_util/verify/interactor.py

@@ -521,12 +521,12 @@ def stage_data_async(
         metadata = test_data.get("metadata", {})
         if not hasattr(metadata, "items"):
             raise Exception(f"Invalid metadata description found for input [{fname}] - [{metadata}]")
-        for name, value in test_data.get("metadata", {}).items():
-            tool_input[f"files_metadata|{name}"] = value
+        for metadata_name, value in test_data.get("metadata", {}).items():
+            tool_input[f"files_metadata|{metadata_name}"] = value
 
         composite_data = test_data["composite_data"]
+        files = {}
         if composite_data:
-            files = {}
             for i, file_name in enumerate(composite_data):
                 if force_path_paste:
                     file_path = self.test_data_path(tool_id, file_name, tool_version=tool_version)
@@ -548,24 +548,21 @@ def stage_data_async(
             location = self._ensure_valid_location_in(test_data)
             if fname:
                 file_name = self.test_data_path(tool_id, fname, tool_version=tool_version)
-                file_name_exists = os.path.exists(f"{file_name}")
-            upload_from_location = not file_name_exists and location is not None
-            name = os.path.basename(location if upload_from_location else fname)
-            tool_input.update(
-                {
-                    "files_0|NAME": name,
-                    "files_0|type": "upload_dataset",
-                }
-            )
-            files = {}
-            if upload_from_location:
-                tool_input.update({"files_0|url_paste": location})
-            elif force_path_paste:
-                file_name = self.test_data_path(tool_id, fname, tool_version=tool_version)
-                tool_input.update({"files_0|url_paste": f"file://{file_name}"})
+                file_name_exists = os.path.exists(file_name)
+            tool_input["files_0|type"] = "upload_dataset"
+            if not file_name_exists and location is not None:
+                name = location
+                tool_input["files_0|url_paste"] = location
             else:
-                file_content = self.test_data_download(tool_id, fname, is_output=False, tool_version=tool_version)
-                files = {"files_0|file_data": file_content}
+                name = fname
+                if force_path_paste:
+                    file_name = self.test_data_path(tool_id, fname, tool_version=tool_version)
+                    tool_input["files_0|url_paste"] = f"file://{file_name}"
+                else:
+                    file_content = self.test_data_download(tool_id, fname, is_output=False, tool_version=tool_version)
+                    files["files_0|file_data"] = file_content
+            name = os.path.basename(name)
+            tool_input["files_0|NAME"] = name
         submit_response_object = self.__submit_tool(
             history_id, "upload1", tool_input, extra_data={"type": "upload_dataset"}, files=files
         )
@@ -587,8 +584,7 @@ def stage_data_async(
 
     def _ensure_valid_location_in(self, test_data: dict) -> Optional[str]:
         location: Optional[str] = test_data.get("location")
-        has_valid_location = location and util.is_url(location)
-        if location and not has_valid_location:
+        if location and not util.is_url(location):
             raise ValueError(f"Invalid `location` URL: `{location}`")
         return location
 

lib/galaxy/tools/__init__.py

@@ -737,7 +737,7 @@ def decode(self, values, tool, app):
         Restore the state from a string
         """
         values = safe_loads(values) or {}
-        self.page = values.pop("__page__") if "__page__" in values else None
+        self.page = values.pop("__page__") if "__page__" in values else 0
         self.rerun_remap_job_id = values.pop("__rerun_remap_job_id__") if "__rerun_remap_job_id__" in values else None
         self.inputs = params_from_strings(tool.inputs, values, app, ignore_errors=True)
 

lib/galaxy/webapps/galaxy/api/library_datasets.py

@@ -487,9 +487,7 @@ def load(self, trans, payload=None, **kwd):
                     os.path.realpath(path),
                 )
                 raise exceptions.RequestParameterInvalidException("The given path is invalid.")
-            if trans.app.config.user_library_import_check_permissions and not full_path_permission_for_user(
-                full_dir, path, username
-            ):
+            if username is not None and not full_path_permission_for_user(full_dir, path, username):
                 log.error(
                     "User attempted to import a path that resolves to a path outside of their import dir: "
                     "%s -> %s and cannot be read by them.",

lib/galaxy/webapps/galaxy/api/users.py

@@ -395,17 +395,16 @@ def remove_favorite(
         user = self.service.get_user(trans, user_id)
         favorites = json.loads(user.preferences["favorites"]) if "favorites" in user.preferences else {}
         if object_type.value == "tools":
-            if "tools" in favorites:
-                favorite_tools = favorites["tools"]
-                if object_id in favorite_tools:
-                    del favorite_tools[favorite_tools.index(object_id)]
-                    favorites["tools"] = favorite_tools
-                    user.preferences["favorites"] = json.dumps(favorites)
-                    with transaction(trans.sa_session):
-                        trans.sa_session.commit()
-                else:
-                    raise exceptions.ObjectNotFound("Given object is not in the list of favorites")
-        return favorites
+            favorite_tools = favorites.get("tools", [])
+            if object_id in favorite_tools:
+                del favorite_tools[favorite_tools.index(object_id)]
+                favorites["tools"] = favorite_tools
+                user.preferences["favorites"] = json.dumps(favorites)
+                with transaction(trans.sa_session):
+                    trans.sa_session.commit()
+            else:
+                raise exceptions.ObjectNotFound("Given object is not in the list of favorites")
+        return FavoriteObjectsSummary.model_validate(favorites)
 
     @router.put(
         "/api/users/{user_id}/favorites/{object_type}",
@@ -428,17 +427,14 @@ def set_favorite(
                 raise exceptions.ObjectNotFound(f"Could not find tool with id '{tool_id}'.")
             if not tool.allow_user_access(user):
                 raise exceptions.AuthenticationFailed(f"Access denied for tool with id '{tool_id}'.")
-            if "tools" in favorites:
-                favorite_tools = favorites["tools"]
-            else:
-                favorite_tools = []
+            favorite_tools = favorites.get("tools", [])
             if tool_id not in favorite_tools:
                 favorite_tools.append(tool_id)
                 favorites["tools"] = favorite_tools
                 user.preferences["favorites"] = json.dumps(favorites)
                 with transaction(trans.sa_session):
                     trans.sa_session.commit()
-        return favorites
+        return FavoriteObjectsSummary.model_validate(favorites)
 
     @router.put(
         "/api/users/{user_id}/theme/{theme}",

lib/galaxy/webapps/galaxy/services/library_contents.py

@@ -163,7 +163,7 @@ def create(
                     with tempfile.NamedTemporaryFile(
                         dir=trans.app.config.new_file_path, prefix="upload_file_data_", delete=False
                     ) as dest:
-                        shutil.copyfileobj(upload_file.file, dest)  # type: ignore[misc]  # https://github.com/python/mypy/issues/15031
+                        shutil.copyfileobj(upload_file.file, dest)
                     upload_file.file.close()
                     upload_files.append(dict(filename=upload_file.filename, local_filename=dest.name))
             payload.upload_files = upload_files

lib/galaxy/webapps/galaxy/services/tools.py

@@ -71,7 +71,7 @@ def create_fetch(
                 with tempfile.NamedTemporaryFile(
                     dir=trans.app.config.new_file_path, prefix="upload_file_data_", delete=False
                 ) as dest:
-                    shutil.copyfileobj(upload_file.file, dest)  # type: ignore[misc]  # https://github.com/python/mypy/issues/15031
+                    shutil.copyfileobj(upload_file.file, dest)
                     util.umask_fix_perms(dest.name, trans.app.config.umask, 0o0666)
                 upload_file.file.close()
                 files_payload[f"files_{i}|file_data"] = FilesPayload(

lib/tool_shed/webapp/api2/repositories.py

@@ -497,7 +497,7 @@ async def create_changeset_revision(
                 dir=trans.app.config.new_file_path, prefix="upload_file_data_", delete=False
             ) as dest:
                 upload_file_like: IO[bytes] = the_file.file
-                shutil.copyfileobj(upload_file_like, dest)  # type: ignore[misc] # https://github.com/python/mypy/issues/15031
+                shutil.copyfileobj(upload_file_like, dest)
             the_file.file.close()
             filename = dest.name
             try: