More data access tests, some refactoring and cleanup

lib/galaxy/managers/groups.py

@@ -11,10 +11,10 @@
     RequestParameterInvalidException,
 )
 from galaxy.managers.context import ProvidesAppContext
-from galaxy.managers.roles import get_roles_by_ids
-from galaxy.managers.users import get_users_by_ids
 from galaxy.model import Group
 from galaxy.model.base import transaction
+from galaxy.model.db.role import get_roles_by_ids
+from galaxy.model.db.user import get_users_by_ids
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.fields import Security
 from galaxy.schema.groups import GroupCreatePayload

lib/galaxy/managers/libraries.py

@@ -10,15 +10,6 @@
     Tuple,
 )
 
-from sqlalchemy import (
-    asc,
-    false,
-    func,
-    not_,
-    or_,
-    select,
-    true,
-)
 from sqlalchemy.exc import (
     MultipleResultsFound,
     NoResultFound,
@@ -29,10 +20,16 @@
 from galaxy.managers.folders import FolderManager
 from galaxy.model import (
     Library,
-    LibraryPermissions,
     Role,
 )
 from galaxy.model.base import transaction
+from galaxy.model.db.library import (
+    get_libraries_by_name,
+    get_libraries_for_admins,
+    get_libraries_for_nonadmins,
+    get_library_ids,
+    get_library_permissions_by_role,
+)
 from galaxy.util import (
     pretty_print_time_interval,
     unicodify,
@@ -60,7 +57,7 @@ def get(self, trans, decoded_library_id: int, check_accessible: bool = True) ->
         :rtype:     galaxy.model.Library
         """
         try:
-            library = get_library(trans.sa_session, decoded_library_id)
+            library = trans.sa_session.get(Library, decoded_library_id)
         except MultipleResultsFound:
             raise exceptions.InconsistentDatabase("Multiple libraries found with the same id.")
         except NoResultFound:
@@ -358,52 +355,8 @@ def get_containing_library_from_library_dataset(trans, library_dataset) -> Optio
     while folder.parent:
         folder = folder.parent
     # We have folder set to the library's root folder, which has the same name as the library
-    stmt = select(Library).where(Library.deleted == false()).where(Library.name == folder.name)
-    for library in trans.sa_session.scalars(stmt):
+    for library in get_libraries_by_name(trans.sa_session, folder.name):
         # Just to double-check
         if library.root_folder == folder:
             return library
     return None
-
-
-def get_library(session, library_id):
-    stmt = select(Library).where(Library.id == library_id)
-    return session.execute(stmt).scalar_one()
-
-
-def get_library_ids(session, library_access_action):
-    stmt = select(LibraryPermissions.library_id).where(LibraryPermissions.action == library_access_action).distinct()
-    return session.scalars(stmt)
-
-
-def get_library_permissions_by_role(session, role_ids):
-    stmt = select(LibraryPermissions).where(LibraryPermissions.role_id.in_(role_ids))
-    return session.scalars(stmt)
-
-
-def get_libraries_for_admins(session, deleted):
-    stmt = select(Library)
-    if deleted is None:
-        #  Flag is not specified, do not filter on it.
-        pass
-    elif deleted:
-        stmt = stmt.where(Library.deleted == true())
-    else:
-        stmt = stmt.where(Library.deleted == false())
-    stmt = stmt.order_by(asc(func.lower(Library.name)))
-    return session.scalars(stmt)
-
-
-def get_libraries_for_nonadmins(session, restricted_library_ids, accessible_restricted_library_ids):
-    stmt = (
-        select(Library)
-        .where(Library.deleted == false())
-        .where(
-            or_(
-                not_(Library.id.in_(restricted_library_ids)),
-                Library.id.in_(accessible_restricted_library_ids),
-            )
-        )
-    )
-    stmt = stmt.order_by(asc(func.lower(Library.name)))
-    return session.scalars(stmt)

lib/galaxy/managers/roles.py

@@ -26,7 +26,6 @@
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import Role
 from galaxy.model.base import transaction
-from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.schema.schema import RoleDefinitionModel
 from galaxy.util import unicodify
 
@@ -162,8 +161,3 @@ def undelete(self, trans: ProvidesUserContext, role: model.Role) -> model.Role:
         with transaction(trans.sa_session):
             trans.sa_session.commit()
         return role
-
-
-def get_roles_by_ids(session: galaxy_scoped_session, role_ids):
-    stmt = select(Role).where(Role.id.in_(role_ids))
-    return session.scalars(stmt).all()

lib/galaxy/managers/users.py

@@ -45,7 +45,10 @@
     UserQuotaUsage,
 )
 from galaxy.model.base import transaction
-from galaxy.model.scoped_session import galaxy_scoped_session
+from galaxy.model.db.user import (
+    get_user_by_email,
+    get_user_by_username,
+)
 from galaxy.security.validate_user_input import (
     VALID_EMAIL_RE,
     validate_email,
@@ -872,25 +875,3 @@ def _add_parsers(self):
         )
 
         self.fn_filter_parsers.update({})
-
-
-def get_users_by_ids(session: galaxy_scoped_session, user_ids):
-    stmt = select(User).where(User.id.in_(user_ids))
-    return session.scalars(stmt).all()
-
-
-# The get_user_by_email and get_user_by_username functions may be called from
-# the tool_shed app, which has its own User model, which is different from
-# galaxy.model.User. In that case, the tool_shed user model should be passed as
-# the model_class argument.
-def get_user_by_email(session, email: str, model_class=User, case_sensitive=True):
-    filter_clause = model_class.email == email
-    if not case_sensitive:
-        filter_clause = func.lower(model_class.email) == func.lower(email)
-    stmt = select(model_class).where(filter_clause).limit(1)
-    return session.scalars(stmt).first()
-
-
-def get_user_by_username(session, username: str, model_class=User):
-    stmt = select(model_class).filter(model_class.username == username).limit(1)
-    return session.scalars(stmt).first()

lib/galaxy/model/db/library.py

@@ -0,0 +1,57 @@
+from sqlalchemy import (
+    asc,
+    false,
+    func,
+    not_,
+    or_,
+    select,
+    true,
+)
+
+from galaxy.model import (
+    Library,
+    LibraryPermissions,
+)
+
+
+def get_library_ids(session, library_access_action):
+    stmt = select(LibraryPermissions.library_id).where(LibraryPermissions.action == library_access_action).distinct()
+    return session.scalars(stmt)
+
+
+def get_library_permissions_by_role(session, role_ids):
+    stmt = select(LibraryPermissions).where(LibraryPermissions.role_id.in_(role_ids))
+    return session.scalars(stmt)
+
+
+def get_libraries_for_admins(session, deleted):
+    stmt = select(Library)
+    if deleted is None:
+        #  Flag is not specified, do not filter on it.
+        pass
+    elif deleted:
+        stmt = stmt.where(Library.deleted == true())
+    else:
+        stmt = stmt.where(Library.deleted == false())
+    stmt = stmt.order_by(asc(func.lower(Library.name)))
+    return session.scalars(stmt)
+
+
+def get_libraries_for_nonadmins(session, restricted_library_ids, accessible_restricted_library_ids):
+    stmt = (
+        select(Library)
+        .where(Library.deleted == false())
+        .where(
+            or_(
+                not_(Library.id.in_(restricted_library_ids)),
+                Library.id.in_(accessible_restricted_library_ids),
+            )
+        )
+    )
+    stmt = stmt.order_by(asc(func.lower(Library.name)))
+    return session.scalars(stmt)
+
+
+def get_libraries_by_name(session, name):
+    stmt = select(Library).where(Library.deleted == false()).where(Library.name == name)
+    return session.scalars(stmt)

lib/galaxy/model/db/role.py

@@ -0,0 +1,43 @@
+from sqlalchemy import (
+    and_,
+    false,
+    select,
+)
+
+from galaxy.model import (
+    Role,
+    UserRoleAssociation,
+)
+from galaxy.model.scoped_session import galaxy_scoped_session
+
+
+def get_npns_roles(session):
+    """
+    non-private, non-sharing roles
+    """
+    stmt = (
+        select(Role)
+        .where(and_(Role.deleted == false(), Role.type != Role.types.PRIVATE, Role.type != Role.types.SHARING))
+        .order_by(Role.name)
+    )
+    return session.scalars(stmt)
+
+
+def get_private_user_role(user, session):
+    stmt = (
+        select(Role)
+        .where(
+            and_(
+                UserRoleAssociation.user_id == user.id,
+                Role.id == UserRoleAssociation.role_id,
+                Role.type == Role.types.PRIVATE,
+            )
+        )
+        .distinct()
+    )
+    return session.execute(stmt).scalar_one_or_none()
+
+
+def get_roles_by_ids(session: galaxy_scoped_session, role_ids):
+    stmt = select(Role).where(Role.id.in_(role_ids))
+    return session.scalars(stmt).all()

lib/galaxy/model/db/user.py

@@ -0,0 +1,66 @@
+from typing import Optional
+
+from sqlalchemy import (
+    false,
+    func,
+    or_,
+    select,
+    true,
+)
+
+from galaxy.model import User
+from galaxy.model.scoped_session import galaxy_scoped_session
+
+
+def get_users_by_ids(session: galaxy_scoped_session, user_ids):
+    stmt = select(User).where(User.id.in_(user_ids))
+    return session.scalars(stmt).all()
+
+
+# The get_user_by_email and get_user_by_username functions may be called from
+# the tool_shed app, which has its own User model, which is different from
+# galaxy.model.User. In that case, the tool_shed user model should be passed as
+# the model_class argument.
+def get_user_by_email(session, email: str, model_class=User, case_sensitive=True):
+    filter_clause = model_class.email == email
+    if not case_sensitive:
+        filter_clause = func.lower(model_class.email) == func.lower(email)
+    stmt = select(model_class).where(filter_clause).limit(1)
+    return session.scalars(stmt).first()
+
+
+def get_user_by_username(session, username: str, model_class=User):
+    stmt = select(model_class).filter(model_class.username == username).limit(1)
+    return session.scalars(stmt).first()
+
+
+def get_users_for_index(
+    session,
+    deleted: bool,
+    f_email: Optional[str] = None,
+    f_name: Optional[str] = None,
+    f_any: Optional[str] = None,
+    is_admin: bool = False,
+    expose_user_email: bool = False,
+    expose_user_name: bool = False,
+):
+    stmt = select(User)
+    if f_email and (is_admin or expose_user_email):
+        stmt = stmt.where(User.email.like(f"%{f_email}%"))
+    if f_name and (is_admin or expose_user_name):
+        stmt = stmt.where(User.username.like(f"%{f_name}%"))
+    if f_any:
+        if is_admin:
+            stmt = stmt.where(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+        else:
+            if expose_user_email and expose_user_name:
+                stmt = stmt.where(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+            elif expose_user_email:
+                stmt = stmt.where(User.email.like(f"%{f_any}%"))
+            elif expose_user_name:
+                stmt = stmt.where(User.username.like(f"%{f_any}%"))
+    if deleted:
+        stmt = stmt.where(User.deleted == true())
+    else:
+        stmt = stmt.where(User.deleted == false())
+    return session.scalars(stmt).all()