Annotate DatasetAssociationManager as generic type

bound to ``DatasetInstance``.

Also:
- Drop redundant ``HDAManager.is_accessible()``.
- Small refactorings.

lib/galaxy/managers/datasets.py

@@ -8,6 +8,7 @@
 from typing import (
     Any,
     Dict,
+    Generic,
     List,
     Optional,
     Type,
@@ -32,6 +33,7 @@
     Dataset,
     DatasetHash,
     DatasetInstance,
+    HistoryDatasetAssociation,
 )
 from galaxy.model.base import transaction
 from galaxy.schema.tasks import (
@@ -124,7 +126,7 @@ def is_accessible(self, item: Any, user: Optional[model.User], **kwargs) -> bool
 
     def has_access_permission(self, dataset, user):
         """
-        Return T/F if the user has role-based access to the dataset.
+        Whether the user has role-based access to the dataset.
         """
         roles = user.all_roles_exploiting_cache() if user else []
         return self.app.security_agent.can_access_dataset(roles, dataset)
@@ -319,12 +321,15 @@ def serialize_permissions(self, item, key, user=None, **context):
         return permissions
 
 
-# ============================================================================= AKA DatasetInstanceManager
+U = TypeVar("U", bound=DatasetInstance)
+
+
 class DatasetAssociationManager(
     base.ModelManager[DatasetInstance],
     secured.AccessibleManagerMixin,
     secured.OwnableManagerMixin,
     deletable.PurgableManagerMixin,
+    Generic[U],
 ):
     """
     DatasetAssociation/DatasetInstances are intended to be working
@@ -342,14 +347,14 @@ def __init__(self, app: MinimalManagerApp):
         super().__init__(app)
         self.dataset_manager = DatasetManager(app)
 
-    def is_accessible(self, item, user: Optional[model.User], **kwargs: Any) -> bool:
+    def is_accessible(self, item: U, user: Optional[model.User], **kwargs: Any) -> bool:
         """
         Is this DA accessible to `user`?
         """
         # defer to the dataset
         return self.dataset_manager.is_accessible(item.dataset, user, **kwargs)
 
-    def delete(self, item, flush: bool = True, stop_job: bool = False, **kwargs):
+    def delete(self, item: U, flush: bool = True, stop_job: bool = False, **kwargs):
         """
         Marks this dataset association as deleted.
         If `stop_job` is True, will stop the creating job if all other outputs are deleted.
@@ -359,7 +364,7 @@ def delete(self, item, flush: bool = True, stop_job: bool = False, **kwargs):
             self.stop_creating_job(item, flush=flush)
         return item
 
-    def purge(self, item, flush=True, **kwargs):
+    def purge(self, item: U, flush=True, **kwargs):
         """
         Purge this DatasetInstance and the dataset underlying it.
         """
@@ -385,7 +390,7 @@ def by_user(self, user):
         raise exceptions.NotImplemented("Abstract Method")
 
     # .... associated job
-    def creating_job(self, dataset_assoc):
+    def creating_job(self, dataset_assoc: U):
         """
         Return the `Job` that created this dataset or None if not found.
         """
@@ -397,7 +402,7 @@ def creating_job(self, dataset_assoc):
             break
         return job
 
-    def stop_creating_job(self, dataset_assoc, flush=False):
+    def stop_creating_job(self, dataset_assoc: U, flush=False):
         """
         Stops an dataset_assoc's creating job if all the job's other outputs are deleted.
         """
@@ -424,21 +429,21 @@ def stop_creating_job(self, dataset_assoc, flush=False):
                     return True
         return False
 
-    def is_composite(self, dataset_assoc):
+    def is_composite(self, dataset_assoc: U):
         """
         Return True if this hda/ldda is a composite type dataset.
 
         .. note:: see also (whereever we keep information on composite datatypes?)
         """
         return dataset_assoc.extension in self.app.datatypes_registry.get_composite_extensions()
 
-    def extra_files(self, dataset_assoc):
+    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 []
         return glob.glob(os.path.join(dataset_assoc.dataset.extra_files_path, "*"))
 
-    def serialize_dataset_association_roles(self, dataset_assoc):
+    def serialize_dataset_association_roles(self, dataset_assoc: U):
         if hasattr(dataset_assoc, "library_dataset_dataset_association"):
             library_dataset = dataset_assoc
             dataset = library_dataset.library_dataset_dataset_association.dataset
@@ -469,13 +474,16 @@ def serialize_dataset_association_roles(self, dataset_assoc):
             rval["modify_item_roles"] = modify_item_role_list
         return rval
 
-    def ensure_dataset_on_disk(self, trans, dataset):
+    def ensure_dataset_on_disk(self, trans, dataset: U):
         # Not a guarantee data is really present, but excludes a lot of expected cases
         if not dataset.dataset:
             raise exceptions.InternalServerError("Item has no associated dataset.")
         if dataset.purged or dataset.dataset.purged:
             raise exceptions.ItemDeletionException("The dataset you are attempting to view has been purged.")
-        elif dataset.deleted and not (trans.user_is_admin or self.is_owner(dataset, trans.get_user())):
+        elif dataset.deleted and not (
+            trans.user_is_admin
+            or (isinstance(dataset, HistoryDatasetAssociation) and self.is_owner(dataset, trans.get_user()))  # type: ignore[arg-type]
+        ):
             raise exceptions.ItemDeletionException("The dataset you are attempting to view has been deleted.")
         elif dataset.state == Dataset.states.UPLOAD:
             raise exceptions.Conflict("Please wait until this dataset finishes uploading before attempting to view it.")
@@ -500,7 +508,7 @@ def ensure_dataset_on_disk(self, trans, dataset):
             if not self.app.object_store.exists(dataset.dataset):
                 raise exceptions.RequestParameterInvalidException("The dataset is in error and has no data.")
 
-    def ensure_can_change_datatype(self, dataset: DatasetInstance, raiseException: bool = True) -> bool:
+    def ensure_can_change_datatype(self, dataset: U, raiseException: bool = True) -> bool:
         if not dataset.datatype.is_datatype_change_allowed():
             if not raiseException:
                 return False
@@ -509,7 +517,7 @@ def ensure_can_change_datatype(self, dataset: DatasetInstance, raiseException: b
             )
         return True
 
-    def ensure_can_set_metadata(self, dataset: DatasetInstance, raiseException: bool = True) -> bool:
+    def ensure_can_set_metadata(self, dataset: U, raiseException: bool = True) -> bool:
         if not dataset.ok_to_edit_metadata():
             if not raiseException:
                 return False
@@ -518,7 +526,7 @@ def ensure_can_set_metadata(self, dataset: DatasetInstance, raiseException: bool
             )
         return True
 
-    def detect_datatype(self, trans, dataset_assoc: DatasetInstance):
+    def detect_datatype(self, trans, dataset_assoc: U):
         """Sniff and assign the datatype to a given dataset association (ldda or hda)"""
         session = self.session()
         self.ensure_can_change_datatype(dataset_assoc)
@@ -531,9 +539,7 @@ def detect_datatype(self, trans, dataset_assoc: DatasetInstance):
             session.commit()
         self.set_metadata(trans, dataset_assoc)
 
-    def set_metadata(
-        self, trans, dataset_assoc: DatasetInstance, overwrite: bool = False, validate: bool = True
-    ) -> None:
+    def set_metadata(self, trans, dataset_assoc: U, overwrite: bool = False, validate: bool = True) -> None:
         """Trigger a job that detects and sets metadata on a given dataset association (ldda or hda)"""
         self.ensure_can_set_metadata(dataset_assoc)
         if overwrite:
@@ -554,7 +560,7 @@ def overwrite_metadata(self, data):
                 if spec.get("default"):
                     setattr(data.metadata, name, spec.unwrap(spec.get("default")))
 
-    def update_permissions(self, trans, dataset_assoc, **kwd):
+    def update_permissions(self, trans, dataset_assoc: U, **kwd):
         action = kwd.get("action", "set_permissions")
         if action not in ["remove_restrictions", "make_private", "set_permissions"]:
             raise exceptions.RequestParameterInvalidException(
@@ -608,7 +614,7 @@ def parameters_roles_or_none(role_type):
 
             self._set_permissions(trans, dataset_assoc, role_ids_dict)
 
-    def _set_permissions(self, trans, dataset_assoc, roles_dict):
+    def _set_permissions(self, trans, dataset_assoc: U, roles_dict):
         raise exceptions.NotImplemented()
 
 

lib/galaxy/managers/hdas.py

@@ -46,6 +46,7 @@
 )
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.model import (
+    HistoryDatasetAssociation,
     Job,
     JobStateHistory,
     JobToOutputDatasetAssociation,
@@ -81,15 +82,15 @@ class HistoryDatasetAssociationNoHistoryException(Exception):
 
 
 class HDAManager(
-    datasets.DatasetAssociationManager,
+    datasets.DatasetAssociationManager[HistoryDatasetAssociation],
     secured.OwnableManagerMixin,
     annotatable.AnnotatableManagerMixin,
 ):
     """
     Interface/service object for interacting with HDAs.
     """
 
-    model_class = model.HistoryDatasetAssociation
+    model_class = HistoryDatasetAssociation
     foreign_key_name = "history_dataset_association"
 
     tag_assoc = model.HistoryDatasetAssociationTagAssociation
@@ -118,22 +119,11 @@ def get_owned_ids(self, object_ids, history=None):
         return self.list(filters=filters)
 
     # .... security and permissions
-    def is_accessible(self, item: model.HistoryDatasetAssociation, user: Optional[model.User], **kwargs: Any) -> bool:
-        """
-        Override to allow owners (those that own the associated history).
-        """
-        # this, apparently, is not True:
-        #   if I have a copy of a dataset and anyone who manages permissions on it revokes my access
-        #   I can not access that dataset even if it's in my history
-        # if self.is_owner( hda, user, **kwargs ):
-        #     return True
-        return super().is_accessible(item, user, **kwargs)
-
     def is_owner(self, item, user: Optional[model.User], current_history=None, **kwargs: Any) -> bool:
         """
         Use history to see if current user owns HDA.
         """
-        if not isinstance(item, model.HistoryDatasetAssociation):
+        if not isinstance(item, HistoryDatasetAssociation):
             raise TypeError('"item" must be of type HistoryDatasetAssociation.')
         if self.user_manager.is_admin(user, trans=kwargs.get("trans", None)):
             return True
@@ -144,15 +134,13 @@ def is_owner(self, item, user: Optional[model.User], current_history=None, **kwa
         # TODO: some dup here with historyManager.is_owner but prevents circ import
         # TODO: awkward kwarg (which is my new band name); this may not belong here - move to controller?
         if self.user_manager.is_anonymous(user):
-            if current_history and history == current_history:
-                return True
-            return False
+            return current_history is not None and history == current_history
         return history.user == user
 
     # .... create and copy
     def create(
         self, flush: bool = True, history=None, dataset=None, *args: Any, **kwargs: Any
-    ) -> model.HistoryDatasetAssociation:
+    ) -> HistoryDatasetAssociation:
         """
         Create a new hda optionally passing in it's history and dataset.
 
@@ -161,9 +149,7 @@ def create(
         """
         if not dataset:
             kwargs["create_dataset"] = True
-        hda = model.HistoryDatasetAssociation(
-            history=history, dataset=dataset, sa_session=self.app.model.context, **kwargs
-        )
+        hda = HistoryDatasetAssociation(history=history, dataset=dataset, sa_session=self.app.model.context, **kwargs)
 
         if history:
             history.add_dataset(hda, set_hid=("hid" not in kwargs))
@@ -204,11 +190,11 @@ def materialize(self, request: MaterializeDatasetInstanceTaskRequest, in_place:
 
     def copy(
         self, item: Any, history=None, hide_copy: bool = False, flush: bool = True, **kwargs: Any
-    ) -> model.HistoryDatasetAssociation:
+    ) -> HistoryDatasetAssociation:
         """
         Copy hda, including annotation and tags, add to history and return the given HDA.
         """
-        if not isinstance(item, model.HistoryDatasetAssociation):
+        if not isinstance(item, HistoryDatasetAssociation):
             raise TypeError()
         hda = item
         copy = hda.copy(
@@ -353,7 +339,7 @@ def _set_permissions(self, trans, hda, role_ids_dict):
 
 def dereference_input(
     trans: ProvidesHistoryContext, data_request: DataRequestUri, history: Optional[model.History] = None
-) -> model.HistoryDatasetAssociation:
+) -> HistoryDatasetAssociation:
     target_history = history or trans.history
     hda = dereference_to_model(trans.sa_session, trans.user, target_history, data_request)
     permissions = trans.app.security_agent.history_get_default_permissions(target_history)
@@ -368,24 +354,24 @@ def __init__(self, hda_manager: HDAManager, dataset_manager: datasets.DatasetMan
         self.hda_manager = hda_manager
         self.dataset_manager = dataset_manager
         self.sort_map = {
-            StoredItemOrderBy.NAME_ASC: asc(model.HistoryDatasetAssociation.name),
-            StoredItemOrderBy.NAME_DSC: desc(model.HistoryDatasetAssociation.name),
+            StoredItemOrderBy.NAME_ASC: asc(HistoryDatasetAssociation.name),
+            StoredItemOrderBy.NAME_DSC: desc(HistoryDatasetAssociation.name),
             StoredItemOrderBy.SIZE_ASC: nulls_first(asc(model.Dataset.total_size)),
             StoredItemOrderBy.SIZE_DSC: nulls_last(desc(model.Dataset.total_size)),
-            StoredItemOrderBy.UPDATE_TIME_ASC: asc(model.HistoryDatasetAssociation.update_time),
-            StoredItemOrderBy.UPDATE_TIME_DSC: desc(model.HistoryDatasetAssociation.update_time),
+            StoredItemOrderBy.UPDATE_TIME_ASC: asc(HistoryDatasetAssociation.update_time),
+            StoredItemOrderBy.UPDATE_TIME_DSC: desc(HistoryDatasetAssociation.update_time),
         }
 
     def get_discarded_summary(self, user: model.User) -> CleanableItemsSummary:
         stmt = (
-            select(func.sum(model.Dataset.total_size), func.count(model.HistoryDatasetAssociation.id))
-            .select_from(model.HistoryDatasetAssociation)
-            .join(model.Dataset, model.HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
-            .join(model.History, model.HistoryDatasetAssociation.table.c.history_id == model.History.id)
+            select(func.sum(model.Dataset.total_size), func.count(HistoryDatasetAssociation.id))
+            .select_from(HistoryDatasetAssociation)
+            .join(model.Dataset, HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
+            .join(model.History, HistoryDatasetAssociation.table.c.history_id == model.History.id)
             .where(
                 and_(
-                    model.HistoryDatasetAssociation.deleted == true(),
-                    model.HistoryDatasetAssociation.purged == false(),  # type:ignore[arg-type]
+                    HistoryDatasetAssociation.deleted == true(),
+                    HistoryDatasetAssociation.purged == false(),  # type:ignore[arg-type]
                     model.History.user_id == user.id,
                 )
             )
@@ -404,18 +390,18 @@ def get_discarded(
     ) -> List[StoredItem]:
         stmt = (
             select(
-                model.HistoryDatasetAssociation.id,
-                model.HistoryDatasetAssociation.name,
-                model.HistoryDatasetAssociation.update_time,
+                HistoryDatasetAssociation.id,
+                HistoryDatasetAssociation.name,
+                HistoryDatasetAssociation.update_time,
                 model.Dataset.total_size,
             )
-            .select_from(model.HistoryDatasetAssociation)
-            .join(model.Dataset, model.HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
-            .join(model.History, model.HistoryDatasetAssociation.table.c.history_id == model.History.id)
+            .select_from(HistoryDatasetAssociation)
+            .join(model.Dataset, HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
+            .join(model.History, HistoryDatasetAssociation.table.c.history_id == model.History.id)
             .where(
                 and_(
-                    model.HistoryDatasetAssociation.deleted == true(),
-                    model.HistoryDatasetAssociation.purged == false(),  # type:ignore[arg-type]
+                    HistoryDatasetAssociation.deleted == true(),
+                    HistoryDatasetAssociation.purged == false(),  # type:ignore[arg-type]
                     model.History.user_id == user.id,
                 )
             )
@@ -441,7 +427,7 @@ def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCle
 
         for hda_id in item_ids:
             try:
-                hda: model.HistoryDatasetAssociation = self.hda_manager.get_owned(hda_id, user)
+                hda: HistoryDatasetAssociation = self.hda_manager.get_owned(hda_id, user)
                 hda.deleted = True
                 quota_amount = int(hda.quota_amount(user))
                 hda.purge_usage_from_quota(user, hda.dataset.quota_source_info)
@@ -649,7 +635,7 @@ def serialize_display_apps(self, item, key, trans=None, **context):
         """
         hda = item
         display_apps: List[Dict[str, Any]] = []
-        if hda.state == model.HistoryDatasetAssociation.states.OK and not hda.deleted:
+        if hda.state == HistoryDatasetAssociation.states.OK and not hda.deleted:
             for display_app in hda.get_display_applications(trans).values():
                 app_links = []
                 for link_app in display_app.links.values():
@@ -673,7 +659,7 @@ def serialize_old_display_applications(self, item, key, trans=None, **context):
         display_apps: List[Dict[str, Any]] = []
         if (
             self.app.config.enable_old_display_applications
-            and hda.state == model.HistoryDatasetAssociation.states.OK
+            and hda.state == HistoryDatasetAssociation.states.OK
             and not hda.deleted
         ):
             display_link_fn = hda.datatype.get_display_links
@@ -760,7 +746,7 @@ class HDAFilterParser(
     datasets.DatasetAssociationFilterParser, taggable.TaggableFilterMixin, annotatable.AnnotatableFilterMixin
 ):
     model_manager_class = HDAManager
-    model_class = model.HistoryDatasetAssociation
+    model_class = HistoryDatasetAssociation
 
     def _add_parsers(self):
         super()._add_parsers()

lib/galaxy/managers/histories.py

@@ -263,9 +263,7 @@ def is_owner(
         """
         # anon users are only allowed to view their current history
         if self.user_manager.is_anonymous(user):
-            if current_history and item == current_history:
-                return True
-            return False
+            return current_history is not None and item == current_history
         return super().is_owner(item, user)
 
     # TODO: possibly to sharable or base