From 456616aa7a5bbf6948a7c8d1374e568c78a3296c Mon Sep 17 00:00:00 2001 From: John Davis Date: Mon, 23 Oct 2023 16:53:30 -0400 Subject: [PATCH] Replace ModelManager.munge_lists with new util function Also, drop _munge_filters and use util function directly. --- lib/galaxy/managers/base.py | 29 ++--------------------------- lib/galaxy/managers/histories.py | 3 ++- lib/galaxy/managers/sharable.py | 11 +++++++---- lib/galaxy/managers/users.py | 5 +++-- 4 files changed, 14 insertions(+), 34 deletions(-) diff --git a/lib/galaxy/managers/base.py b/lib/galaxy/managers/base.py index 76107688e4a..1b413310b6c 100644 --- a/lib/galaxy/managers/base.py +++ b/lib/galaxy/managers/base.py @@ -68,6 +68,7 @@ from galaxy.structured_app import ( BasicSharedApp, MinimalManagerApp, ) +from galaxy.util import munge_lists from galaxy.web import url_for as gx_url_for log = logging.getLogger(__name__) @@ -194,24 +195,6 @@ def get_object(trans, id, class_name, check_ownership=False, check_accessible=Fa # ============================================================================= -def munge_lists(listA, listB): - """ - Combine two lists into a single list. - - (While allowing them to be None, non-lists, or lists.) - """ - # TODO: there's nothing specifically filter or model-related here - move to util - if listA is None: - return listB - if listB is None: - return listA - if not isinstance(listA, list): - listA = [listA] - if not isinstance(listB, list): - listB = [listB] - return listA + listB - - U = TypeVar("U", bound=model._HasTable) @@ -288,14 +271,6 @@ class ModelManager(Generic[U]): query = query.filter(filter) return query - def _munge_filters(self, filtersA, filtersB): - """ - Combine two lists into a single list. - - (While allowing them to be None, non-lists, or lists.) - """ - return munge_lists(filtersA, filtersB) - # .... order, limit, and offset def _apply_order_by(self, query: Query, order_by) -> Query: """ @@ -487,7 +462,7 @@ class ModelManager(Generic[U]): if not ids: return [] ids_filter = parsed_filter("orm", self.model_class.__table__.c.id.in_(ids)) - found = self.list(filters=self._munge_filters(ids_filter, filters), **kwargs) + found = self.list(filters=munge_lists(ids_filter, filters), **kwargs) # TODO: this does not order by the original 'ids' array # ...could use get (supposedly since found are in the session, the db won't be hit twice) diff --git a/lib/galaxy/managers/histories.py b/lib/galaxy/managers/histories.py index 2f89fb1bed8..22166843f34 100644 --- a/lib/galaxy/managers/histories.py +++ b/lib/galaxy/managers/histories.py @@ -60,6 +60,7 @@ from galaxy.schema.storage_cleaner import ( ) from galaxy.security.validate_user_input import validate_preferred_object_store_id from galaxy.structured_app import MinimalManagerApp +from galaxy.util import munge_lists log = logging.getLogger(__name__) @@ -135,7 +136,7 @@ class HistoryManager(sharable.SharableModelManager, deletable.PurgableManagerMix if self.user_manager.is_anonymous(user): return None if (not current_history or current_history.deleted) else current_history desc_update_time = desc(self.model_class.update_time) - filters = self._munge_filters(filters, self.model_class.user_id == user.id) + filters = munge_lists(filters, self.model_class.user_id == user.id) # TODO: normalize this return value return self.query(filters=filters, order_by=desc_update_time, limit=1, **kwargs).first() diff --git a/lib/galaxy/managers/sharable.py b/lib/galaxy/managers/sharable.py index 92499743b3d..6eadd811028 100644 --- a/lib/galaxy/managers/sharable.py +++ b/lib/galaxy/managers/sharable.py @@ -45,7 +45,10 @@ from galaxy.schema.schema import ( SharingOptions, ) from galaxy.structured_app import MinimalManagerApp -from galaxy.util import ready_name_for_url +from galaxy.util import ( + munge_lists, + ready_name_for_url, +) from galaxy.util.hash_util import md5_hash_str if TYPE_CHECKING: @@ -83,7 +86,7 @@ class SharableModelManager( `user`. """ user_filter = self.model_class.table.c.user_id == user.id - filters = self._munge_filters(user_filter, kwargs.get("filters", None)) + filters = munge_lists(user_filter, kwargs.get("filters", None)) return self.list(filters=filters, **kwargs) # .... owned/accessible interfaces @@ -154,7 +157,7 @@ class SharableModelManager( Return a query for all published items. """ published_filter = self.model_class.table.c.published == true() - filters = self._munge_filters(published_filter, filters) + filters = munge_lists(published_filter, filters) return self.query(filters=filters, **kwargs) def list_published(self, filters=None, **kwargs): @@ -162,7 +165,7 @@ class SharableModelManager( Return a list of all published items. """ published_filter = self.model_class.table.c.published == true() - filters = self._munge_filters(published_filter, filters) + filters = munge_lists(published_filter, filters) return self.list(filters=filters, **kwargs) # .... user sharing diff --git a/lib/galaxy/managers/users.py b/lib/galaxy/managers/users.py index 8d08d767f35..f2aca660ec8 100644 --- a/lib/galaxy/managers/users.py +++ b/lib/galaxy/managers/users.py @@ -54,6 +54,7 @@ from galaxy.structured_app import ( BasicSharedApp, MinimalManagerApp, ) +from galaxy.util import munge_lists from galaxy.util.hash_util import new_secure_hash_v2 from galaxy.web import url_for @@ -267,7 +268,7 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin): """ Find a user by their email. """ - filters = self._munge_filters(self.model_class.email == email, filters) + filters = munge_lists(self.model_class.email == email, filters) try: # TODO: use one_or_none return super().one(filters=filters, **kwargs) @@ -321,7 +322,7 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin): Return a list of admin Users. """ admin_emails = self.app.config.admin_users_list - filters = self._munge_filters(self.model_class.email.in_(admin_emails), filters) + filters = munge_lists(self.model_class.email.in_(admin_emails), filters) return super().list(filters=filters, **kwargs) def error_unless_admin(self, user, msg="Administrators only", **kwargs):