mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Replace ModelManager.munge_lists with new util function
Also, drop _munge_filters and use util function directly.
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user