Fix UP031 errors

reported by ruff 0.8.0 .

cron/build_chrom_db.py

@@ -47,7 +47,7 @@ def getchrominfo(url, db):
         if len(fields) > 1 and len(fields[0]) > 0 and int(fields[1]) > 0:
             yield [fields[0], fields[1]]
         else:
-            raise Exception("Problem parsing line %d '%s' in page '%s'" % (i, line, page))
+            raise Exception(f"Problem parsing line {i} '{line}' in page '{page}'")
 
 
 if __name__ == "__main__":

lib/galaxy/app.py

@@ -551,7 +551,7 @@ def _wait_for_database(self, url):
                 database_exists(url)
                 break
             except Exception:
-                log.info("Waiting for database: attempt %d of %d" % (i, attempts))
+                log.info("Waiting for database: attempt %d of %d", i, attempts)
                 time.sleep(pause)
 
     @property

lib/galaxy_test/api/test_jobs.py

@@ -1129,10 +1129,9 @@ def _search(self, payload, expected_search_count=1):
             if search_count == expected_search_count:
                 break
             time.sleep(1)
-        assert search_count == expected_search_count, "expected to find %d jobs, got %d jobs" % (
-            expected_search_count,
-            search_count,
-        )
+        assert (
+            search_count == expected_search_count
+        ), f"expected to find {expected_search_count} jobs, got {search_count} jobs"
         return search_count
 
     def _search_count(self, search_payload):

lib/galaxy_test/api/test_workflow_extraction.py

@@ -5,6 +5,7 @@
     dumps,
     loads,
 )
+from typing import Optional
 
 from galaxy_test.base.populators import (
     skip_without_tool,
@@ -611,11 +612,11 @@ def _extract_and_download_workflow(self, history_id: str, **extract_payload):
         downloaded_workflow = download_response.json()
         return downloaded_workflow
 
-    def _get_steps_of_type(self, downloaded_workflow, type, expected_len=None):
+    def _get_steps_of_type(self, downloaded_workflow, type: str, expected_len: Optional[int] = None):
         steps = [s for s in downloaded_workflow["steps"].values() if s["type"] == type]
         if expected_len is not None:
             n = len(steps)
-            assert n == expected_len, "Expected %d steps of type %s, found %d" % (expected_len, type, n)
+            assert n == expected_len, f"Expected {expected_len} steps of type {type}, found {n}"
         return sorted(steps, key=operator.itemgetter("id"))
 
     def __job_id(self, history_id, dataset_id):
@@ -645,10 +646,10 @@ def _run_tool_get_collection_and_job_id(self, history_id: str, tool_id, inputs):
     def __check_workflow(
         self,
         workflow,
-        step_count=None,
-        verify_connected=False,
-        data_input_count=None,
-        data_collection_input_count=None,
+        step_count: Optional[int] = None,
+        verify_connected: bool = False,
+        data_input_count: Optional[int] = None,
+        data_collection_input_count: Optional[int] = None,
         tool_ids=None,
     ):
         steps = workflow["steps"]

lib/galaxy_test/selenium/framework.py

@@ -233,15 +233,15 @@ def func_wrapper(self, *args, **kwds):
 class TestSnapshot:
     __test__ = False  # Prevent pytest from discovering this class (issue #12071)
 
-    def __init__(self, driver, index, description):
+    def __init__(self, driver, index: int, description: str):
         self.screenshot_binary = driver.get_screenshot_as_png()
         self.description = description
         self.index = index
         self.exc = traceback.format_exc()
         self.stack = traceback.format_stack()
 
     def write_to_error_directory(self, write_file_func):
-        prefix = "%d-%s" % (self.index, self.description)
+        prefix = f"{self.index}-{self.description}"
         write_file_func(f"{prefix}-screenshot.png", self.screenshot_binary, raw=True)
         write_file_func(f"{prefix}-traceback.txt", self.exc)
         write_file_func(f"{prefix}-stack.txt", str(self.stack))
@@ -331,7 +331,7 @@ def setup_with_driver(self):
     def tear_down_selenium(self):
         self.tear_down_driver()
 
-    def snapshot(self, description):
+    def snapshot(self, description: str):
         """Create a debug snapshot (DOM, screenshot, etc...) that is written out on tool failure.
 
         This information will be automatically written to a per-test directory created for all
@@ -371,7 +371,7 @@ def _screenshot_path(self, label, extension=".png"):
         copy = 1
         while os.path.exists(target):
             # Maybe previously a test re-run - keep the original.
-            target = os.path.join(GALAXY_TEST_SCREENSHOTS_DIRECTORY, "%s-%d%s" % (label, copy, extension))
+            target = os.path.join(GALAXY_TEST_SCREENSHOTS_DIRECTORY, f"{label}-{copy}{extension}")
             copy += 1
 
         return target

lib/galaxy_test/selenium/test_uploads.py

@@ -16,7 +16,7 @@ def test_upload_file(self):
 
         self.history_panel_wait_for_hid_ok(1)
         history_count = len(self.history_contents())
-        assert history_count == 1, "Incorrect number of items in history - expected 1, found %d" % history_count
+        assert history_count == 1, f"Incorrect number of items in history - expected 1, found {history_count}"
 
         self.history_panel_click_item_title(hid=1, wait=True)
         self.assert_item_summary_includes(1, "28 lines")
@@ -28,7 +28,7 @@ def test_upload_pasted_content(self):
 
         self.history_panel_wait_for_hid_ok(1)
         history_count = len(self.history_contents())
-        assert history_count == 1, "Incorrect number of items in history - expected 1, found %d" % history_count
+        assert history_count == 1, f"Incorrect number of items in history - expected 1, found {history_count}"
 
     @selenium_test
     def test_upload_pasted_url_content(self):
@@ -37,7 +37,7 @@ def test_upload_pasted_url_content(self):
 
         self.history_panel_wait_for_hid_ok(1)
         history_count = len(self.history_contents())
-        assert history_count == 1, "Incorrect number of items in history - expected 1, found %d" % history_count
+        assert history_count == 1, f"Incorrect number of items in history - expected 1, found {history_count}"
 
     @selenium_test
     def test_upload_composite_dataset_pasted_data(self):
@@ -46,7 +46,7 @@ def test_upload_composite_dataset_pasted_data(self):
 
         self.history_panel_wait_for_hid_ok(1)
         history_count = len(self.history_contents())
-        assert history_count == 1, "Incorrect number of items in history - expected 1, found %d" % history_count
+        assert history_count == 1, f"Incorrect number of items in history - expected 1, found {history_count}"
 
         self.history_panel_click_item_title(hid=1, wait=True)
         self.history_panel_item_view_dataset_details(1)
@@ -62,7 +62,7 @@ def test_upload_simplest(self):
         self.history_panel_wait_for_hid_ok(1)
         history_contents = self.history_contents()
         history_count = len(history_contents)
-        assert history_count == 1, "Incorrect number of items in history - expected 1, found %d" % history_count
+        assert history_count == 1, f"Incorrect number of items in history - expected 1, found {history_count}"
 
         hda = history_contents[0]
         assert hda["name"] == "1.sam", hda

scripts/api/common.py

@@ -92,7 +92,7 @@ def display(api_key, url, return_formatted=True):
                 # All collection members should have a name in the response.
                 # url is optional
                 if "url" in i:
-                    print("#%d: %s" % (n + 1, i.pop("url")))
+                    print("#{}: {}".format(n + 1, i.pop("url")))
                 if "name" in i:
                     print(f"  name: {i.pop('name')}")
                 try:
@@ -102,7 +102,7 @@ def display(api_key, url, return_formatted=True):
                     for item in i:
                         print(item)
         print("")
-        print("%d element(s) in collection" % len(r))
+        print(f"{len(r)} element(s) in collection")
     elif isinstance(r, dict):
         # Response is an element as defined in the REST style.
         print("Member Information")

scripts/apply_tags.py

@@ -48,7 +48,7 @@ def find_dataset_parents_update_tags(self, history, job, history_id):
         count_datasets_updated = 0
         # get all datasets belonging to a history
         all_datasets = history.show_history(history_id, contents=True)
-        print("Total datasets: %d. Updating their tags may take a while..." % len(all_datasets))
+        print(f"Total datasets: {len(all_datasets)}. Updating their tags may take a while...")
         for dataset in all_datasets:
             try:
                 if not dataset["deleted"] and dataset["state"] == "ok":
@@ -91,7 +91,7 @@ def find_dataset_parents_update_tags(self, history, job, history_id):
                 )
                 if is_updated:
                     count_datasets_updated += 1
-        print("Tags of %d datasets updated" % count_datasets_updated)
+        print(f"Tags of {count_datasets_updated} datasets updated")
 
     def collect_parent_ids(self, datasets_inheritance_chain):
         """
@@ -158,4 +158,4 @@ def propagate_tags(self, history, current_history_id, parent_datasets_ids, datas
     history_tags = ApplyTagsHistory(sys.argv[1], sys.argv[2], history_id)
     history_tags.read_galaxy_history()
     end_time = time.time()
-    print("Program finished in %d seconds" % int(end_time - start_time))
+    print(f"Program finished in {int(end_time - start_time)} seconds")

scripts/build_toolbox.py

@@ -55,7 +55,7 @@ def add(self, toolelement, toolboxpositionelement):
             sectionorder = self.sectionorders[section]
 
         # Sortorder: add intelligent mix to the front
-        self.tools[("%05d-%s" % (sectionorder, section), label, order, section)].append(toolelement)
+        self.tools[(f"{sectionorder:05d}-{section}", label, order, section)].append(toolelement)
 
     def addElementsTo(self, rootelement):
         toolkeys = list(self.tools.keys())
@@ -77,7 +77,7 @@ def addElementsTo(self, rootelement):
                 currentlabel = ""
                 if section:
                     sectionnumber += 1
-                    attrib = {"name": section, "id": "section%d" % sectionnumber}
+                    attrib = {"name": section, "id": f"section{sectionnumber}"}
                     sectionelement = ET.Element("section", attrib)
                     rootelement.append(sectionelement)
                     currentelement = sectionelement
@@ -90,7 +90,7 @@ def addElementsTo(self, rootelement):
                 currentlabel = label
                 if label:
                     labelnumber += 1
-                    attrib = {"text": label, "id": "label%d" % labelnumber}
+                    attrib = {"text": label, "id": f"label{labelnumber}"}
                     labelelement = ET.Element("label", attrib)
                     currentelement.append(labelelement)
 

scripts/cleanup_datasets/admin_cleanup_datasets.py

@@ -171,7 +171,7 @@ def main():
     now = strftime("%Y-%m-%d %H:%M:%S")
 
     print("##########################################")
-    print("\n# %s - Handling stuff older than %i days" % (now, args.days))
+    print(f"\n# {now} - Handling stuff older than {args.days} days")
 
     if args.info_only:
         print("# Displaying info only ( --info_only )\n")
@@ -260,15 +260,15 @@ def administrative_delete_datasets(
                     # Mark the HistoryDatasetAssociation as deleted
                     hda.deleted = True
                     app.sa_session.add(hda)
-                    print("Marked HistoryDatasetAssociation id %d as deleted" % hda.id)
+                    print(f"Marked HistoryDatasetAssociation id {hda.id} as deleted")
                 session = app.sa_session()
                 with transaction(session):
                     session.commit()
 
     emailtemplate = Template(filename=template_file)
     for email, dataset_list in user_notifications.items():
         msgtext = emailtemplate.render(email=email, datasets=dataset_list, cutoff=cutoff_days)
-        subject = "Galaxy Server Cleanup - %d datasets DELETED" % len(dataset_list)
+        subject = f"Galaxy Server Cleanup - {len(dataset_list)} datasets DELETED"
         fromaddr = config.email_from
         print()
         print(f"From: {fromaddr}")
@@ -281,7 +281,7 @@ def administrative_delete_datasets(
 
     stop = time.time()
     print()
-    print("Marked %d dataset instances as deleted" % deleted_instance_count)
+    print(f"Marked {deleted_instance_count} dataset instances as deleted")
     print("Total elapsed time: ", stop - start)
     print("##########################################")
 

scripts/cleanup_datasets/pgcleanup.py

@@ -409,7 +409,7 @@ def recalculate_disk_usage(self):
                     new_args[key] = val
                 self._update(sql, new_args, add_event=False)
 
-            self.log.info("recalculate_disk_usage user_id %i" % user_id)
+            self.log.info("recalculate_disk_usage user_id %i", user_id)
 
 
 class RemovesMetadataFiles(RemovesObjects):
@@ -1317,7 +1317,8 @@ def _dry_run_event(self):
         else:
             log.info(
                 "Not executing event creation (increments sequence even when rolling back), using an old "
-                "event ID (%i) for dry run" % max_id
+                "event ID (%i) for dry run",
+                max_id,
             )
         return max_id
 

scripts/drmaa_external_killer.py

@@ -12,7 +12,7 @@
 import drmaa
 
 
-def validate_paramters():
+def validate_parameters():
     if len(sys.argv) < 3:
         sys.stderr.write(f"usage: {sys.argv[0]} [job ID] [user uid]\n")
         exit(1)
@@ -22,15 +22,15 @@ def validate_paramters():
     return jobID, uid
 
 
-def set_user(uid):
+def set_user(uid: int):
     try:
         gid = pwd.getpwuid(uid).pw_gid
         os.setgid(gid)
         os.setuid(uid)
     except OSError as e:
         if e.errno == errno.EPERM:
             sys.stderr.write(
-                "error: setuid(%d) failed: permission denied. Did you setup 'sudo' correctly for this script?\n" % uid
+                f"error: setuid({uid}) failed: permission denied. Did you setup 'sudo' correctly for this script?\n"
             )
             exit(1)
         else:
@@ -48,7 +48,7 @@ def set_user(uid):
 
 
 def main():
-    jobID, uid = validate_paramters()
+    jobID, uid = validate_parameters()
     set_user(uid)
     s = drmaa.Session()
     s.initialize()

scripts/drmaa_external_runner.py

@@ -41,7 +41,7 @@ def valid_numeric_userid(userid):
     try:
         pwd.getpwuid(uid)
     except KeyError:
-        sys.stderr.write("error: User-ID (%d) is not valid.\n" % uid)
+        sys.stderr.write(f"error: User-ID ({uid}) is not valid.\n")
         exit(1)
     return True
 
@@ -63,7 +63,7 @@ def json_file_exists(json_filename):
     return True
 
 
-def validate_paramters():
+def validate_parameters():
     assign_all_groups = False
     if "--assign_all_groups" in sys.argv:
         assign_all_groups = True
@@ -88,7 +88,7 @@ def validate_paramters():
     return uid, json_filename, assign_all_groups
 
 
-def set_user(uid, assign_all_groups):
+def set_user(uid: int, assign_all_groups: bool):
     try:
         # Get user's default group and set it to current process to make sure
         # file permissions are inherited correctly
@@ -108,7 +108,7 @@ def set_user(uid, assign_all_groups):
     except OSError as e:
         if e.errno == errno.EPERM:
             sys.stderr.write(
-                "error: setuid(%d) failed: permission denied. Did you setup 'sudo' correctly for this script?\n" % uid
+                f"error: setuid({uid}) failed: permission denied. Did you setup 'sudo' correctly for this script?\n"
             )
             exit(1)
         else:
@@ -128,7 +128,7 @@ def set_user(uid, assign_all_groups):
 
 
 def main():
-    userid, json_filename, assign_all_groups = validate_paramters()
+    userid, json_filename, assign_all_groups = validate_parameters()
     # load JSON job template data
     json_file_exists(json_filename)
     with open(json_filename) as f:

scripts/metagenomics/convert_title.py

@@ -18,7 +18,7 @@
             if len_seq > 0:
                 if gi is None:
                     raise Exception("The first sequence does not have an header.")
-                print(">%s_%d" % (gi, len_seq))
+                print(f">{gi}_{len_seq}")
                 print("\n".join(seq))
             title = line
             fields = title.split("|")
@@ -33,10 +33,10 @@
             seq.append(line)
             len_seq += len(line)
     if len_seq > 0:
-        print(">%s_%d" % (gi, len_seq))
+        print(f">{gi}_{len_seq}")
         print("\n".join(seq))
 
     print(
-        "Unable to find gi number for %d sequences, the title is replaced as giunknown" % (invalid_lines),
+        f"Unable to find gi number for {invalid_lines} sequences, the title is replaced as giunknown",
         file=sys.stderr,
     )

scripts/runtime_stats.py

@@ -205,14 +205,14 @@ def query(
     if debug:
         print("Executed:")
         print(cur.query)
-    print("Query returned %d rows" % cur.rowcount)
+    print(f"Query returned {cur.rowcount} rows")
 
     if source == "metrics":
         times = numpy.array([r[0] for r in cur if r[0]])
     elif source == "history":
         times = numpy.array([r[0].total_seconds() for r in cur if r[0]])
 
-    print("Collected %d times" % times.size)
+    print(f"Collected {times.size} times")
 
     if times.size == 0:
         return