Type annotation fixes for mypy 1.14.0

lib/galaxy/datatypes/tabular.py

@@ -807,7 +807,7 @@ def set_meta(
         >>> set_datatypes_registry(example_datatype_registry_for_sample())
         >>> fname = get_test_fname( 'sam_with_header.sam' )
         >>> samds = Dataset(external_filename=fname)
-        >>> hda = hist.add_dataset(HistoryDatasetAssociation(id=1, extension='sam', create_dataset=True, sa_session=sa_session, dataset=samds))
+        >>> hda = hist.add_dataset(HistoryDatasetAssociation(id=1, extension='sam', sa_session=sa_session, dataset=samds))
         >>> Sam().set_meta(hda)
         >>> hda.metadata.comment_lines
         2

lib/galaxy/dependencies/pinned-typecheck-requirements.txt

@@ -6,7 +6,7 @@ botocore-stubs==1.35.81
 cffi==1.17.1 ; platform_python_implementation != 'PyPy'
 cryptography==44.0.0
 lxml-stubs==0.5.1
-mypy==1.13.0
+mypy==1.14.0
 mypy-boto3-s3==1.35.76.post1
 mypy-extensions==1.0.0
 pycparser==2.22 ; platform_python_implementation != 'PyPy'

lib/galaxy/files/unittest_utils/__init__.py

@@ -1,6 +1,9 @@
 import os
 import tempfile
-from typing import Tuple
+from typing import (
+    Optional,
+    Tuple,
+)
 
 from galaxy.files import (
     ConfiguredFileSources,
@@ -10,7 +13,7 @@
 
 
 class TestConfiguredFileSources(ConfiguredFileSources):
-    def __init__(self, file_sources_config: FileSourcePluginsConfig, conf_dict: dict, test_root: str):
+    def __init__(self, file_sources_config: FileSourcePluginsConfig, conf_dict: dict, test_root: Optional[str]):
         super().__init__(file_sources_config, ConfiguredFileSourcesConf(conf_dict=conf_dict))
         self.test_root = test_root
 

lib/galaxy/jobs/runners/tasks.py

@@ -235,7 +235,7 @@ def _stop_pid(self, pid, job_id):
                 log.warning(
                     "_stop_pid(): %s: Got errno %s when attempting to signal %d to PID %d: %s",
                     job_id,
-                    errno.errorcode[e.errno],
+                    errno.errorcode[e.errno] if e.errno is not None else None,
                     sig,
                     pid,
                     e.strerror,

lib/galaxy/jobs/runners/util/process_groups.py

@@ -17,7 +17,7 @@ def check_pg(pgid):
         else:
             log.warning(
                 "check_pg(): Got errno %s when checking process group %d: %s",
-                errno.errorcode[e.errno],
+                errno.errorcode[e.errno] if e.errno is not None else None,
                 pgid,
                 e.strerror,
             )
@@ -37,7 +37,7 @@ def kill_pg(pgid):
                 return
             log.warning(
                 "Got errno %s when sending signal %d to process group %d: %s",
-                errno.errorcode[e.errno],
+                errno.errorcode[e.errno] if e.errno is not None else None,
                 sig,
                 pgid,
                 e.strerror,

lib/galaxy/managers/datasets.py

@@ -8,7 +8,6 @@
 from typing import (
     Any,
     Dict,
-    Generic,
     List,
     Optional,
     Type,
@@ -327,11 +326,10 @@ def serialize_permissions(self, item, key, user=None, **context):
 
 
 class DatasetAssociationManager(
-    base.ModelManager[DatasetInstance],
+    base.ModelManager[U],
     secured.AccessibleManagerMixin,
     secured.OwnableManagerMixin,
     deletable.PurgableManagerMixin,
-    Generic[U],
 ):
     """
     DatasetAssociation/DatasetInstances are intended to be working
@@ -384,6 +382,7 @@ def purge(self, item: U, flush=True, **kwargs):
         self.stop_creating_job(item, flush=True)
 
         # more importantly, purge underlying dataset as well
+        assert item.dataset
         if item.dataset.user_can_purge:
             self.dataset_manager.purge(item.dataset, flush=flush, **kwargs)
         return item
@@ -443,6 +442,7 @@ def extra_files(self, dataset_assoc: U):
         """Return a list of file paths for composite files, an empty list otherwise."""
         if not self.is_composite(dataset_assoc):
             return []
+        assert dataset_assoc.dataset
         return glob.glob(os.path.join(dataset_assoc.dataset.extra_files_path, "*"))
 
     def serialize_dataset_association_roles(self, dataset_assoc: U):

lib/galaxy/managers/hdas.py

@@ -230,24 +230,25 @@ def purge(self, item, flush=True, **kwargs):
         else:
             self._purge(item, flush=flush)
 
-    def _purge(self, hda, flush=True):
+    def _purge(self, hda: HistoryDatasetAssociation, flush: bool = True):
         """
         Purge this HDA and the dataset underlying it.
         """
         user = hda.history.user or None
-        quota_amount_reduction = 0
         if user:
+            # Need to calculate this before purging
             quota_amount_reduction = hda.quota_amount(user)
         super().purge(hda, flush=flush)
         # decrease the user's space used
-        quota_source_info = hda.dataset.quota_source_info
-        if quota_amount_reduction and quota_source_info.use:
-            user.adjust_total_disk_usage(-quota_amount_reduction, quota_source_info.label)
-            # TODO: don't flush above if we're going to re-flush here
-            session = object_session(user)
-            assert session
-            with transaction(session):
-                session.commit()
+        if user:
+            quota_source_info = hda.dataset.quota_source_info
+            if quota_amount_reduction and quota_source_info.use:
+                user.adjust_total_disk_usage(-quota_amount_reduction, quota_source_info.label)
+                # TODO: don't flush above if we're going to re-flush here
+                session = object_session(user)
+                assert session
+                with transaction(session):
+                    session.commit()
 
     # .... states
     def error_if_uploading(self, hda):

lib/galaxy/model/__init__.py

@@ -4687,7 +4687,6 @@ def __init__(
         extension=None,
         dbkey=None,
         metadata=None,
-        history=None,
         dataset=None,
         deleted=False,
         designation=None,

lib/galaxy/model/dereference.py

@@ -31,6 +31,7 @@ def dereference_to_model(sa_session, user, history, data_request_uri: DataReques
         hash_object.hash_value = dataset_hash.hash_value
         hashes.append(hash_object)
     dataset_source.hashes = hashes
+    assert hda.dataset
     hda.dataset.sources = [dataset_source]
     transform: List[TransformAction] = []
     if data_request_uri.space_to_tab:

lib/galaxy/model/store/__init__.py

@@ -24,6 +24,7 @@
     Iterable,
     Iterator,
     List,
+    Literal,
     Optional,
     Set,
     Tuple,
@@ -2987,9 +2988,7 @@ def get_export_store_factory(
 
 
 def tar_export_directory(export_directory: StrPath, out_file: StrPath, gzip: bool) -> None:
-    tarfile_mode = "w"
-    if gzip:
-        tarfile_mode += ":gz"
+    tarfile_mode: Literal["w", "w:gz"] = "w:gz" if gzip else "w"
 
     with tarfile.open(out_file, tarfile_mode, dereference=True) as store_archive:
         for export_path in os.listdir(export_directory):

lib/galaxy/tool_util/verify/interactor.py

@@ -1028,7 +1028,7 @@ def prepare_request_params(
                 new_items[key] = dumps(val)
         data.update(new_items)
 
-    kwd = {
+    kwd: Dict[str, Any] = {
         "files": files,
     }
     if headers:

lib/galaxy/util/compression_utils.py

@@ -182,7 +182,7 @@ class CompressedFile:
     def can_decompress(file_path: StrPath) -> bool:
         return tarfile.is_tarfile(file_path) or zipfile.is_zipfile(file_path)
 
-    def __init__(self, file_path: StrPath, mode: str = "r") -> None:
+    def __init__(self, file_path: StrPath, mode: Literal["a", "r", "w", "x"] = "r") -> None:
         file_path_str = str(file_path)
         if tarfile.is_tarfile(file_path):
             self.file_type = "tar"
@@ -336,11 +336,10 @@ def isfile(self, member: ArchiveMemberType) -> bool:
             return True
         return False
 
-    def open_tar(self, filepath: StrPath, mode: str) -> tarfile.TarFile:
+    def open_tar(self, filepath: StrPath, mode: Literal["a", "r", "w", "x"]) -> tarfile.TarFile:
         return tarfile.open(filepath, mode, errorlevel=0)
 
-    def open_zip(self, filepath: StrPath, mode: str) -> zipfile.ZipFile:
-        mode = cast(Literal["a", "r", "w", "x"], mode)
+    def open_zip(self, filepath: StrPath, mode: Literal["a", "r", "w", "x"]) -> zipfile.ZipFile:
         return zipfile.ZipFile(filepath, mode)
 
     def zipfile_ok(self, path_to_archive: StrPath) -> bool:

test/unit/files/test_posix.py

@@ -3,7 +3,6 @@
 from typing import (
     Any,
     Dict,
-    Tuple,
 )
 
 import pytest
@@ -98,6 +97,7 @@ def test_posix_link_security_allowlist():
 def test_posix_link_security_allowlist_write():
     file_sources = _configured_file_sources(include_allowlist=True, writable=True)
     write_from(file_sources, "gxfiles://test1/unsafe_dir/foo", "my test content")
+    assert file_sources.test_root
     with open(os.path.join(file_sources.test_root, "subdir1", "foo")) as f:
         assert f.read() == "my test content"
 
@@ -438,7 +438,7 @@ def test_posix_file_url_allowed_root():
 
 
 def test_posix_file_url_disallowed_root():
-    file_sources, root = _configured_file_sources_with_root(plugin_extra_config={"enforce_symlink_security": False})
+    file_sources, _ = _configured_file_sources_with_root(plugin_extra_config={"enforce_symlink_security": False})
     with tempfile.NamedTemporaryFile(mode="w") as tf:
         tf.write("some content")
         tf.flush()
@@ -478,7 +478,7 @@ def _configured_file_sources_with_root(
     writable=None,
     allow_subdir_creation=True,
     empty_root=False,
-) -> Tuple[TestConfiguredFileSources, str]:
+):
     if empty_root:
         tmp, root = "/", None
     else: