Do not reuse util.listify

Move function to managers.base; rename.
Do not reuse listify; implement standalone function.
Add test case for tuples.
(See PR code review for discussion)
This commit is contained in:
John Davis
2023-10-24 09:40:03 -04:00
parent 456616aa7a
commit b172849313
7 changed files with 52 additions and 42 deletions
+22 -2
View File
@@ -68,7 +68,6 @@ 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__)
@@ -462,7 +461,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=munge_lists(ids_filter, filters), **kwargs)
found = self.list(filters=combine_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)
@@ -1364,3 +1363,24 @@ class StorageCleanerManager(Protocol):
def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCleanupResult:
"""Purges the given list of items by ID. The items must be owned by the user."""
raise NotImplementedError
def combine_lists(listA: Any, listB: Any) -> List:
"""
Combine two lists into a single list.
Arguments can be None, non-lists, or lists. If an argument is None, it will
not be included in the returned list. If both arguments are None, an empty
list will be returned.
"""
def make_list(item):
# Check for None explicitly: __bool__ may be overwritten.
if item is None:
return []
elif isinstance(item, (list, tuple)):
return list(item)
else:
return [item]
return make_list(listA) + make_list(listB)
+2 -2
View File
@@ -37,6 +37,7 @@ from galaxy.managers import (
sharable,
)
from galaxy.managers.base import (
combine_lists,
ModelDeserializingError,
Serializer,
SortableManager,
@@ -60,7 +61,6 @@ 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__)
@@ -136,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 = munge_lists(filters, self.model_class.user_id == user.id)
filters = combine_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()
+5 -7
View File
@@ -34,6 +34,7 @@ from galaxy.managers import (
taggable,
users,
)
from galaxy.managers.base import combine_lists
from galaxy.model import (
User,
UserShareAssociation,
@@ -45,10 +46,7 @@ from galaxy.schema.schema import (
SharingOptions,
)
from galaxy.structured_app import MinimalManagerApp
from galaxy.util import (
munge_lists,
ready_name_for_url,
)
from galaxy.util import ready_name_for_url
from galaxy.util.hash_util import md5_hash_str
if TYPE_CHECKING:
@@ -86,7 +84,7 @@ class SharableModelManager(
`user`.
"""
user_filter = self.model_class.table.c.user_id == user.id
filters = munge_lists(user_filter, kwargs.get("filters", None))
filters = combine_lists(user_filter, kwargs.get("filters", None))
return self.list(filters=filters, **kwargs)
# .... owned/accessible interfaces
@@ -157,7 +155,7 @@ class SharableModelManager(
Return a query for all published items.
"""
published_filter = self.model_class.table.c.published == true()
filters = munge_lists(published_filter, filters)
filters = combine_lists(published_filter, filters)
return self.query(filters=filters, **kwargs)
def list_published(self, filters=None, **kwargs):
@@ -165,7 +163,7 @@ class SharableModelManager(
Return a list of all published items.
"""
published_filter = self.model_class.table.c.published == true()
filters = munge_lists(published_filter, filters)
filters = combine_lists(published_filter, filters)
return self.list(filters=filters, **kwargs)
# .... user sharing
+3 -3
View File
@@ -37,6 +37,7 @@ from galaxy.managers import (
base,
deletable,
)
from galaxy.managers.base import combine_lists
from galaxy.model import (
User,
UserAddress,
@@ -54,7 +55,6 @@ 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
@@ -268,7 +268,7 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin):
"""
Find a user by their email.
"""
filters = munge_lists(self.model_class.email == email, filters)
filters = combine_lists(self.model_class.email == email, filters)
try:
# TODO: use one_or_none
return super().one(filters=filters, **kwargs)
@@ -322,7 +322,7 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin):
Return a list of admin Users.
"""
admin_emails = self.app.config.admin_users_list
filters = munge_lists(self.model_class.email.in_(admin_emails), filters)
filters = combine_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):
-11
View File
@@ -1882,14 +1882,3 @@ def enum_values(enum_class):
Values are in member definition order.
"""
return [value.value for value in enum_class.__members__.values()]
def munge_lists(listA: Any, listB: Any) -> List:
"""
Combine two lists into a single list.
Arguments can be None, non-lists, or lists. If an argument is None, it will
not be included in the returned list. If both arguments are None, an empty
list will be returned.
"""
return listify(listA) + listify(listB)
+20
View File
@@ -0,0 +1,20 @@
from galaxy.managers.base import combine_lists
class NotFalsy:
"""Class requires explicit check for None"""
def __bool__(self):
raise Exception("not implemented")
def test_combine_lists():
foo, bar = NotFalsy(), NotFalsy()
assert combine_lists(foo, None) == [foo]
assert combine_lists(None, foo) == [foo]
assert combine_lists(foo, bar) == [foo, bar]
assert combine_lists([foo, bar], None) == [foo, bar]
assert combine_lists(None, [foo, bar]) == [foo, bar]
assert combine_lists((foo, bar), None) == [foo, bar]
assert combine_lists(None, (foo, bar)) == [foo, bar]
assert combine_lists(None, None) == []
-17
View File
@@ -149,20 +149,3 @@ def test_enum_values():
B = "b"
assert util.enum_values(Stuff) == ["a", "c", "b"]
class NotFalsy:
"""Class requires explicit check for None"""
def __bool__(self):
raise Exception("not implemented")
def test_munge_lists():
foo, bar = NotFalsy(), NotFalsy()
assert util.munge_lists(foo, None) == [foo]
assert util.munge_lists(None, foo) == [foo]
assert util.munge_lists(foo, bar) == [foo, bar]
assert util.munge_lists([foo, bar], None) == [foo, bar]
assert util.munge_lists(None, [foo, bar]) == [foo, bar]
assert util.munge_lists(None, None) == []