Fix UP031 errors - Part 4

lib/galaxy/datatypes/util/maf_utilities.py

@@ -12,6 +12,7 @@
 import tempfile
 from copy import deepcopy
 from errno import EMFILE
+from typing import Dict
 
 import bx.align.maf
 import bx.interval_index_file
@@ -137,26 +138,23 @@ class RegionAlignment:
     DNA_COMPLEMENT = maketrans("ACGTacgt", "TGCAtgca")
     MAX_SEQUENCE_SIZE = sys.maxsize  # Maximum length of sequence allowed
 
-    def __init__(self, size, species=None, temp_file_handler=None):
+    def __init__(self, size: int, species=None, temp_file_handler=None):
         assert (
             size <= self.MAX_SEQUENCE_SIZE
-        ), "Maximum length allowed for an individual sequence has been exceeded (%i > %i)." % (
-            size,
-            self.MAX_SEQUENCE_SIZE,
-        )
+        ), f"Maximum length allowed for an individual sequence has been exceeded ({size} > {self.MAX_SEQUENCE_SIZE})."
         species = species or []
         self.size = size
         if not temp_file_handler:
             temp_file_handler = TempFileHandler()
         self.temp_file_handler = temp_file_handler
-        self.sequences = {}
+        self.sequences: Dict[str, int] = {}
         if not isinstance(species, list):
             species = [species]
         for spec in species:
             self.add_species(spec)
 
     # add a species to the alignment
-    def add_species(self, species):
+    def add_species(self, species: str):
         # make temporary sequence files
         file_index, fh = self.temp_file_handler.get_open_tempfile()
         self.sequences[species] = file_index
@@ -176,13 +174,13 @@ def get_species_names(self, skip=None):
         return names
 
     # returns the sequence for a species
-    def get_sequence(self, species):
+    def get_sequence(self, species: str):
         file_index, fh = self.temp_file_handler.get_open_tempfile(self.sequences[species])
         fh.seek(0)
         return fh.read()
 
     # returns the reverse complement of the sequence for a species
-    def get_sequence_reverse_complement(self, species):
+    def get_sequence_reverse_complement(self, species: str):
         complement = list(self.get_sequence(species).translate(self.DNA_COMPLEMENT))
         complement.reverse()
         return "".join(complement)
@@ -195,9 +193,9 @@ def set_position(self, index, species, base):
 
     # sets a range for a species
 
-    def set_range(self, index, species, bases):
+    def set_range(self, index: int, species: str, bases):
         if index >= self.size or index < 0:
-            raise Exception("Your index (%i) is out of range (0 - %i)." % (index, self.size - 1))
+            raise Exception(f"Your index ({index}) is out of range (0 - {self.size - 1}).")
         if len(bases) == 0:
             raise Exception("A set of genomic positions can only have a positive length.")
         if species not in self.sequences.keys():
@@ -219,7 +217,7 @@ def flush(self, species=None):
 class GenomicRegionAlignment(RegionAlignment):
     def __init__(self, start, end, species=None, temp_file_handler=None):
         species = species or []
-        RegionAlignment.__init__(self, end - start, species, temp_file_handler=temp_file_handler)
+        super().__init__(end - start, species, temp_file_handler=temp_file_handler)
         self.start = start
         self.end = end
 
@@ -734,12 +732,7 @@ def remove_temp_index_file(index_filename):
 
 def get_fasta_header(component, attributes=None, suffix=None):
     attributes = attributes or {}
-    header = ">%s(%s):%i-%i|" % (
-        component.src,
-        component.strand,
-        component.get_forward_strand_start(),
-        component.get_forward_strand_end(),
-    )
+    header = f">{component.src}({component.strand}):{component.get_forward_strand_start()}-{component.get_forward_strand_end()}|"
     for key, value in attributes.items():
         header = f"{header}{key}={value}|"
     if suffix:

lib/galaxy/model/security.py

@@ -367,17 +367,21 @@ def get_actions_for_items(self, trans, action, permission_items):
                     if len(base_result) == len(new_result):
                         common_result = set(base_result).intersection(new_result)
                         if len(common_result) == len(base_result):
-                            log.debug("Match on permissions for id %d" % item.library_dataset_id)
+                            log.debug("Match on permissions for id %d", item.library_dataset_id)
                         # TODO: Fix this failure message:
                         else:
                             log.debug(
-                                "Error: dataset %d; originally: %s; now: %s"
-                                % (item.library_dataset_id, base_result, new_result)
+                                "Error: dataset %d; originally: %s; now: %s",
+                                item.library_dataset_id,
+                                base_result,
+                                new_result,
                             )
                     else:
                         log.debug(
-                            "Error: dataset %d: had %d entries, now %d entries"
-                            % (item.library_dataset_id, len(base_result), len(new_result))
+                            "Error: dataset %d: had %d entries, now %d entries",
+                            item.library_dataset_id,
+                            len(base_result),
+                            len(new_result),
                         )
                 log.debug("get_actions_for_items: Test end")
             except Exception as e:
@@ -431,9 +435,9 @@ def allow_action_on_libitems(self, trans, user_roles, action, items):
             for item in items:
                 orig_value = self.allow_action(user_roles, action, item)
                 if orig_value == ret_allow_action[item.id]:
-                    log.debug("Item %d: success" % item.id)
+                    log.debug("Item %d: success", item.id)
                 else:
-                    log.debug("Item %d: fail: original: %s; new: %s" % (item.id, orig_value, ret_allow_action[item.id]))
+                    log.debug("Item %d: fail: original: %s; new: %s", item.id, orig_value, ret_allow_action[item.id])
             log.debug("allow_action_for_items: test end")
         return ret_allow_action
 
@@ -1394,9 +1398,9 @@ def show_library_item(self, user, roles, library_item, actions_to_check, hidden_
                 if can_show:
                     return True, hidden_folder_ids
                 if hidden_folder_ids:
-                    hidden_folder_ids = "%s,%d" % (hidden_folder_ids, folder.id)
+                    hidden_folder_ids = f"{hidden_folder_ids},{folder.id}"
                 else:
-                    hidden_folder_ids = "%d" % folder.id
+                    hidden_folder_ids = f"{folder.id}"
         return False, hidden_folder_ids
 
     def get_showable_folders(
@@ -1650,9 +1654,9 @@ def check_folder_contents(self, user, roles, folder, hidden_folder_ids=""):
             if can_access:
                 return True, hidden_folder_ids
             if hidden_folder_ids:
-                hidden_folder_ids = "%s,%d" % (hidden_folder_ids, sub_folder.id)
+                hidden_folder_ids = f"{hidden_folder_ids},{sub_folder.id}"
             else:
-                hidden_folder_ids = "%d" % sub_folder.id
+                hidden_folder_ids = f"{sub_folder.id}"
         return False, hidden_folder_ids
 
 
@@ -1690,7 +1694,7 @@ def allow_action(self, addr, action, **kwd):
             if action == self.permitted_actions.DATASET_ACCESS and action.action not in [
                 dp.action for dp in hda.dataset.actions
             ]:
-                log.debug("Allowing access to public dataset with hda: %i." % hda.id)
+                log.debug("Allowing access to public dataset with hda: %d.", hda.id)
                 return True  # dataset has no roles associated with the access permission, thus is already public
             stmt = (
                 select(HistoryDatasetAssociationDisplayAtAuthorization)
@@ -1699,9 +1703,7 @@ def allow_action(self, addr, action, **kwd):
             )
             hdadaa = self.sa_session.scalars(stmt).first()
             if not hdadaa:
-                log.debug(
-                    "Denying access to private dataset with hda: %i.  No hdadaa record for this dataset." % hda.id
-                )
+                log.debug("Denying access to private dataset with hda: %d.  No hdadaa record for this dataset.", hda.id)
                 return False  # no auth
             # We could just look up the reverse of addr, but then we'd also
             # have to verify it with the forward address and special case any
@@ -1719,17 +1721,18 @@ def allow_action(self, addr, action, **kwd):
                     pass  # can't resolve, try next
             else:
                 log.debug(
-                    "Denying access to private dataset with hda: %i.  Remote addr is not a valid server for site: %s."
-                    % (hda.id, hdadaa.site)
+                    "Denying access to private dataset with hda: %d.  Remote addr is not a valid server for site: %s.",
+                    hda.id,
+                    hdadaa.site,
                 )
                 return False  # remote addr is not in the server list
             if (datetime.utcnow() - hdadaa.update_time) > timedelta(seconds=60):
                 log.debug(
-                    "Denying access to private dataset with hda: %i.  Authorization was granted, but has expired."
-                    % hda.id
+                    "Denying access to private dataset with hda: %d.  Authorization was granted, but has expired.",
+                    hda.id,
                 )
                 return False  # not authz'd in the last 60 seconds
-            log.debug("Allowing access to private dataset with hda: %i.  Remote server is: %s." % (hda.id, server))
+            log.debug("Allowing access to private dataset with hda: %d.  Remote server is: %s.", hda.id, server)
             return True
         else:
             raise Exception("The dataset access permission is the only valid permission in the host security agent.")

lib/galaxy/model/store/discover.py

@@ -1050,7 +1050,7 @@ def raw_element_identifiers(self):
         identifiers = []
         i = 0
         while True:
-            key = "identifier_%d" % i
+            key = f"identifier_{i}"
             if key in self.as_dict:
                 identifiers.append(self.as_dict.get(key))
             else:

lib/galaxy/schema/__init__.py

@@ -1,11 +1,11 @@
-import typing
 from copy import deepcopy
 from datetime import datetime
 from enum import Enum
 from typing import (
     Any,
     Callable,
     Dict,
+    Iterable,
     List,
     Optional,
     Tuple,
@@ -136,19 +136,17 @@ def partial_model(
     if exclude is None:
         exclude = []
 
-    @typing.no_type_check  # Mypy doesn't understand pydantic's create_model
     def decorator(model: Type[T]) -> Type[T]:
         def make_optional(field: FieldInfo, default: Any = None) -> Tuple[Any, FieldInfo]:
             new = deepcopy(field)
             new.default = default
-            new.annotation = Optional[field.annotation or Any]
+            new.annotation = Optional[field.annotation or Any]  # type:ignore[assignment]
             return new.annotation, new
 
-        fields = model.model_fields
         if include is None:
-            fields = fields.items()
+            fields: Iterable[Tuple[str, FieldInfo]] = model.model_fields.items()
         else:
-            fields = ((k, v) for k, v in fields.items() if k in include)
+            fields = ((k, v) for k, v in model.model_fields.items() if k in include)
 
         return create_model(
             model.__name__,
@@ -159,6 +157,6 @@ def make_optional(field: FieldInfo, default: Any = None) -> Tuple[Any, FieldInfo
                 for field_name, field_info in fields
                 if exclude is None or field_name not in exclude
             },
-        )
+        )  # type:ignore[call-overload]
 
     return decorator

lib/galaxy/security/validate_user_input.py

@@ -50,13 +50,13 @@ def validate_email_str(email):
     if not (VALID_EMAIL_RE.match(email)):
         return "The format of the email address is not correct."
     elif len(email) > EMAIL_MAX_LEN:
-        return "Email address cannot be more than %d characters in length." % EMAIL_MAX_LEN
+        return f"Email address cannot be more than {EMAIL_MAX_LEN} characters in length."
     return ""
 
 
 def validate_password_str(password):
     if not password or len(password) < PASSWORD_MIN_LEN:
-        return "Use a password of at least %d characters." % PASSWORD_MIN_LEN
+        return f"Use a password of at least {PASSWORD_MIN_LEN} characters."
     return ""
 
 
@@ -65,7 +65,7 @@ def validate_publicname_str(publicname):
     if not publicname:
         return "Public name cannot be empty"
     if len(publicname) > PUBLICNAME_MAX_LEN:
-        return "Public name cannot be more than %d characters in length." % (PUBLICNAME_MAX_LEN)
+        return f"Public name cannot be more than {PUBLICNAME_MAX_LEN} characters in length."
     if not (VALID_PUBLICNAME_RE.match(publicname)):
         return "Public name must contain only lower-case letters, numbers, '.', '_' and '-'."
     return ""

lib/galaxy/selenium/navigates_galaxy.py

@@ -100,7 +100,7 @@ class WAIT_TYPES:
 
 
 class NullTourCallback:
-    def handle_step(self, step, step_index):
+    def handle_step(self, step, step_index: int):
         pass
 
 
@@ -1634,7 +1634,7 @@ def workflow_run_specify_inputs(self, inputs: Dict[str, Any]):
         workflow_run = self.components.workflow_run
         for label, value in inputs.items():
             input_div_element = workflow_run.input_data_div(label=label).wait_for_visible()
-            self.select_set_value(input_div_element, "%d: " % value["hid"])
+            self.select_set_value(input_div_element, "{}: ".format(value["hid"]))
 
     def workflow_run_submit(self):
         self.components.workflow_run.run_workflow.wait_for_and_click()
@@ -2207,7 +2207,7 @@ def assert_message(self, element, contains=None):
     def assert_no_error_message(self):
         self.components._.messages.error.assert_absent_or_hidden()
 
-    def run_tour_step(self, step, step_index, tour_callback):
+    def run_tour_step(self, step, step_index: int, tour_callback):
         element_str = step.get("element", None)
         if element_str is None:
             component = step.get("component", None)

lib/galaxy/selenium/scripts/dump_tour.py

@@ -34,9 +34,9 @@ def __init__(self, driver_wrapper, output):
         self.driver_wrapper = driver_wrapper
         self.output = output
 
-    def handle_step(self, step, step_index):
+    def handle_step(self, step, step_index: int):
         time.sleep(0.5)
-        self.driver_wrapper.driver.save_screenshot("%s/%i.png" % (self.output, step_index))
+        self.driver_wrapper.driver.save_screenshot(f"{self.output}/{step_index}.png")
         time.sleep(0.5)
 
 

lib/galaxy/tool_shed/galaxy_install/metadata/installed_repository_metadata_manager.py

@@ -178,12 +178,12 @@ def reset_metadata_on_selected_repositories(self, user, **kwd):
                 except Exception:
                     log.exception("Error attempting to reset metadata on repository %s", str(repository.name))
                     unsuccessful_count += 1
-            message = "Successfully reset metadata on %d %s.  " % (
+            message = "Successfully reset metadata on {} {}.  ".format(
                 successful_count,
                 inflector.cond_plural(successful_count, "repository"),
             )
             if unsuccessful_count:
-                message += "Error setting metadata on %d %s - see the galaxy log for details.  " % (
+                message += "Error setting metadata on {} {} - see the galaxy log for details.  ".format(
                     unsuccessful_count,
                     inflector.cond_plural(unsuccessful_count, "repository"),
                 )

lib/galaxy/tool_shed/util/repository_util.py

@@ -61,8 +61,8 @@ def check_for_updates(
                 repository_names_not_updated.append(f"<b>{escape(str(repository.name))}</b>")
             if updated:
                 updated_count += 1
-        message = "Checked the status in the tool shed for %d repositories.  " % success_count
-        message += "Updated the tool shed status for %d repositories.  " % updated_count
+        message = f"Checked the status in the tool shed for {success_count} repositories.  "
+        message += f"Updated the tool shed status for {updated_count} repositories.  "
         if repository_names_not_updated:
             message += "Unable to retrieve status from the tool shed for the following repositories:\n"
             message += ", ".join(repository_names_not_updated)

lib/galaxy/tool_util/client/staging.py

@@ -169,10 +169,10 @@ def _attach_file(upload_payload: Dict[str, Any], uri: str, index: int = 0) -> No
                 uri = path_or_uri_to_uri(uri)
                 is_path = uri.startswith("file://")
                 if not is_path or use_path_paste:
-                    upload_payload["inputs"]["files_%d|url_paste" % index] = uri
+                    upload_payload["inputs"][f"files_{index}|url_paste"] = uri
                 else:
                     path = uri[len("file://") :]
-                    upload_payload["__files"]["files_%d|file_data" % index] = self._attach_file(path)
+                    upload_payload["__files"][f"files_{index}|file_data"] = self._attach_file(path)
 
             if isinstance(upload_target, FileUploadTarget):
                 file_path = upload_target.path

lib/galaxy/tool_util/data/__init__.py

@@ -578,10 +578,7 @@ def parse_file_fields(
                     if self.largest_index < len(fields):
                         rval.append(fields)
                     else:
-                        line_error = (
-                            "Line %i in tool data table '%s' is invalid (HINT: '%s' characters must be used to separate fields):\n%s"
-                            % ((i + 1), self.name, separator_char, line)
-                        )
+                        line_error = f"Line {i + 1} in tool data table '{self.name}' is invalid (HINT: '{separator_char}' characters must be used to separate fields):\n{line}"
                         if errors is not None:
                             errors.append(line_error)
                         log.warning(line_error)

lib/galaxy/tool_util/deps/brew_exts.py

@@ -525,7 +525,7 @@ def recipe_cellar_path(cellar_path, recipe, version):
     if revision_paths:
         revisions = (int(x.rsplit("_", 1)[-1]) for x in revision_paths)
         max_revision = max(revisions)
-        recipe_path = "%s_%d" % (recipe_base_path, max_revision)
+        recipe_path = f"{recipe_base_path}_{max_revision}"
     else:
         recipe_path = recipe_base_path
     return recipe_path

lib/galaxy/tool_util/deps/docker_util.py

@@ -160,7 +160,7 @@ def build_docker_run_command(
             euid = os.geteuid()
             egid = os.getgid()
 
-            user = "%d:%d" % (euid, egid)
+            user = f"{euid}:{egid}"
         command_parts.extend(["--user", user])
     full_image = image
     if tag:

lib/galaxy/tool_util/output_checker.py

@@ -102,11 +102,7 @@ def check_output(
                         code_desc = stdio_exit_code.desc
                         if None is code_desc:
                             code_desc = ""
-                        desc = "%s: Exit code %d (%s)" % (
-                            StdioErrorLevel.desc(stdio_exit_code.error_level),
-                            tool_exit_code,
-                            code_desc,
-                        )
+                        desc = f"{StdioErrorLevel.desc(stdio_exit_code.error_level)}: Exit code {tool_exit_code} ({code_desc})"
                         reason = {
                             "type": "exit_code",
                             "desc": desc,