Refactor get_user_by_username, get_user_by_email; use across code base

lib/galaxy/managers/users.py

@@ -359,7 +359,7 @@ def get_user_by_identity(self, identity):
         user = None
         if VALID_EMAIL_RE.match(identity):
             # VALID_PUBLICNAME and VALID_EMAIL do not overlap, so 'identity' here is an email address
-            user = self.session().query(self.model_class).filter(self.model_class.table.c.email == identity).first()
+            user = get_user_by_email(self.session(), identity, self.model_class)
             if not user:
                 # Try a case-insensitive match on the email
                 user = (
@@ -369,7 +369,7 @@ def get_user_by_identity(self, identity):
                     .first()
                 )
         else:
-            user = self.session().query(self.model_class).filter(self.model_class.table.c.username == identity).first()
+            user = get_user_by_username(self.session(), identity, self.model_class)
         return user
 
     # ---- current
@@ -532,7 +532,7 @@ def __get_activation_token(self, trans, email):
         """
         Check for the activation token. Create new activation token and store it in the database if no token found.
         """
-        user = trans.sa_session.query(self.app.model.User).filter(self.app.model.User.table.c.email == email).first()
+        user = get_user_by_email(trans.sa_session, email, self.app.model.User)
         activation_token = user.activation_token
         if activation_token is None:
             activation_token = util.hash_util.new_secure_hash_v2(str(random.getrandbits(256)))
@@ -576,9 +576,7 @@ def send_reset_email(self, trans, payload, **kwd):
                 return "Failed to produce password reset token. User not found."
 
     def get_reset_token(self, trans, email):
-        reset_user = (
-            trans.sa_session.query(self.app.model.User).filter(self.app.model.User.table.c.email == email).first()
-        )
+        reset_user = get_user_by_email(trans.sa_session, email, self.app.model.User)
         if not reset_user and email != email.lower():
             reset_user = (
                 trans.sa_session.query(self.app.model.User)
@@ -622,12 +620,7 @@ def get_or_create_remote_user(self, remote_user_email):
             return None
         if getattr(self.app.config, "normalize_remote_user_email", False):
             remote_user_email = remote_user_email.lower()
-        user = (
-            self.session()
-            .query(self.app.model.User)
-            .filter(self.app.model.User.table.c.email == remote_user_email)
-            .first()
-        )
+        user = get_user_by_email(self.session(), remote_user_email, self.app.model.User)
         if user:
             # GVK: June 29, 2009 - This is to correct the behavior of a previous bug where a private
             # role and default user / history permissions were not set for remote users.  When a
@@ -844,18 +837,20 @@ def _add_parsers(self):
         self.fn_filter_parsers.update({})
 
 
-def get_user_by_username(session, user_class, username):
-    """
-    Get a user from the database by username.
-    (We pass the session and the user_class to accommodate usage from the tool_shed app.)
-    """
-    try:
-        stmt = select(user_class).filter(user_class.username == username)
-        return session.execute(stmt).scalar_one()
-    except Exception:
-        return None
-
-
 def get_users_by_ids(session: 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):
+    stmt = select(model_class).filter(model_class.email == email).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/visualization/genomes.py

@@ -11,6 +11,8 @@
     ObjectNotFound,
     ReferenceDataError,
 )
+from galaxy.managers.users import get_user_by_username
+from galaxy.model import HistoryDatasetAssociation
 from galaxy.structured_app import StructuredApp
 from galaxy.util.bunch import Bunch
 

lib/galaxy/webapps/galaxy/controllers/user.py

@@ -18,6 +18,7 @@
 )
 from galaxy.exceptions import Conflict
 from galaxy.managers import users
+from galaxy.managers.users import get_user_by_email
 from galaxy.security.validate_user_input import (
     validate_email,
     validate_publicname,
@@ -293,9 +294,7 @@ def activate(self, trans, **kwd):
             )
         else:
             # Find the user
-            user = (
-                trans.sa_session.query(trans.app.model.User).filter(trans.app.model.User.table.c.email == email).first()
-            )
+            user = get_user_by_email(trans.sa_session, email)
             if not user:
                 # Probably wrong email address
                 return trans.show_error_message(

lib/galaxy/webapps/galaxy/services/quotas.py

@@ -8,7 +8,7 @@
     get_quotas,
     QuotaManager,
 )
-from galaxy.model.repositories import get_user_by_email
+from galaxy.managers.users import get_user_by_email
 from galaxy.quota._schema import (
     CreateQuotaParams,
     CreateQuotaResult,
@@ -119,7 +119,7 @@ def get_user_id(item):
             try:
                 return trans.security.decode_id(item)
             except Exception:
-                return user_repo.get_by_email(item).id
+                return get_user_by_email(trans.sa_session, item).id
 
         def get_group_id(item):
             try:

lib/galaxy/webapps/reports/controllers/jobs.py

@@ -18,15 +18,14 @@
     and_,
     not_,
     or_,
-    select,
 )
 
 from galaxy import (
     model,
     util,
 )
+from galaxy.managers.users import get_user_by_email
 from galaxy.model import Job
-from galaxy.model.repositories import get_user_by_email
 from galaxy.web.legacy_framework import grids
 from galaxy.webapps.base.controller import (
     BaseUIController,
@@ -1307,8 +1306,7 @@ def get_monitor_id(trans, monitor_email):
     A convenience method to obtain the monitor job id.
     """
     monitor_user_id = None
-    stmt = select(trans.model.User.id).filter(trans.model.User.email == monitor_email).limit(1)
-    monitor_row = trans.sa_session.scalars(stmt).first()
+    monitor_row = get_user_by_email(trans.sa_session, monitor_email)
     if monitor_row is not None:
         monitor_user_id = monitor_row[0]
     return monitor_user_id

lib/tool_shed/test/base/test_db_util.py

@@ -10,6 +10,7 @@
 import galaxy.model
 import galaxy.model.tool_shed_install
 import tool_shed.webapp.model as model
+from galaxy.managers.users import get_user_by_username
 
 log = logging.getLogger("test.tool_shed.test_db_util")
 
@@ -171,10 +172,6 @@ def get_user(email):
     return sa_session().query(model.User).filter(model.User.table.c.email == email).first()
 
 
-def get_user_by_name(username):
-    return sa_session().query(model.User).filter(model.User.table.c.username == username).first()
-
-
 def mark_obj_deleted(obj):
     obj.deleted = True
     sa_session().add(obj)
@@ -190,7 +187,7 @@ def ga_refresh(obj):
 
 
 def get_repository_by_name_and_owner(name, owner_username, return_multiple=False):
-    owner = get_user_by_name(owner_username)
+    owner = get_user_by_username(sa_session(), owner_username, model.User)
     repository = (
         sa_session()
         .query(model.Repository)

lib/tool_shed/webapp/controllers/repository.py

@@ -2293,7 +2293,7 @@ def set_malicious(self, trans, id, ctx_str, **kwd):
     def sharable_owner(self, trans, owner):
         """Support for sharable URL for each repository owner's tools, e.g. http://example.org/view/owner."""
         try:
-            user = get_user_by_username(trans.model.session, trans.model.User, owner)
+            user = get_user_by_username(trans.model.session, owner, trans.model.User)
         except Exception:
             user = None
         if user:
@@ -2321,7 +2321,7 @@ def sharable_repository(self, trans, owner, name):
         else:
             # If the owner is valid, then show all of their repositories.
             try:
-                user = get_user_by_username(trans.model.session, trans.model.User, owner)
+                user = get_user_by_username(trans.model.session, owner, trans.model.User)
             except Exception:
                 user = None
             if user:

test/integration/test_user_preferences.py

@@ -7,8 +7,8 @@
     get,
     put,
 )
-from sqlalchemy import select
 
+from galaxy.managers.users import get_user_by_email
 from galaxy_test.driver import integration_util
 
 TEST_USER_EMAIL = "[email protected]"
@@ -19,8 +19,8 @@ def test_user_theme(self):
         user = self._setup_user(TEST_USER_EMAIL)
         url = self._api_url(f"users/{user['id']}/theme/test_theme", params=dict(key=self.master_api_key))
         app = cast(Any, self._test_driver.app if self._test_driver else None)
-        stmt = select(app.model.User).filter(app.model.User.email == user["email"]).limit(1)
-        db_user = app.model.session.scalars(stmt).first()
+
+        db_user = get_user_by_email(app.model.session, user["email"])
 
         # create some initial data
         put(url)

test/integration/test_vault_extra_prefs.py

@@ -9,8 +9,8 @@
     get,
     put,
 )
-from sqlalchemy import select
 
+from galaxy.managers.users import get_user_by_email
 from galaxy_test.driver import integration_util
 
 TEST_USER_EMAIL = "[email protected]"
@@ -133,5 +133,4 @@ def __url(self, action, user):
         return self._api_url(f"users/{user['id']}/{action}", params=dict(key=self.master_api_key))
 
     def _get_dbuser(self, app, user):
-        stmt = select(app.model.User).filter(app.model.User.email == user["email"]).limit(1)
-        return app.model.session.scalars(stmt).first()
+        return get_user_by_email(app.model.session, user["email"])

test/integration/test_vault_file_source.py

@@ -1,8 +1,7 @@
 import os
 import tempfile
 
-from sqlalchemy import select
-
+from galaxy.managers.users import get_user_by_email
 from galaxy.security.vault import UserVaultWrapper
 from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import DatasetPopulator
@@ -36,7 +35,7 @@ def test_vault_secret_per_user_in_file_source(self):
         app = self._app
 
         with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
-            user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
+            user = get_user_by_email(self._app.model.session, self.USER_1_APP_VAULT_ENTRY)
             user_vault = UserVaultWrapper(app.vault, user)
             # use a valid symlink path so the posix list succeeds
             user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
@@ -48,7 +47,7 @@ def test_vault_secret_per_user_in_file_source(self):
             print(remote_files)
 
         with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
-            user = self._get_user_by_email(self.USER_2_APP_VAULT_ENTRY)
+            user = get_user_by_email(self._app.model.session, self.USER_2_APP_VAULT_ENTRY)
             user_vault = UserVaultWrapper(app.vault, user)
             # use an invalid symlink path so the posix list fails
             user_vault.write_secret("posix/root_path", "/invalid/root")
@@ -69,7 +68,7 @@ def test_vault_secret_per_app_in_file_source(self):
         app.vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
 
         with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
-            user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
+            user = get_user_by_email(self._app.model.session, self.USER_1_APP_VAULT_ENTRY)
             user_vault = UserVaultWrapper(app.vault, user)
             # use a valid symlink path so the posix list succeeds
             user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
@@ -81,7 +80,7 @@ def test_vault_secret_per_app_in_file_source(self):
             print(remote_files)
 
         with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
-            user = self._get_user_by_email(self.USER_2_APP_VAULT_ENTRY)
+            user = get_user_by_email(self._app.model.session, self.USER_2_APP_VAULT_ENTRY)
             user_vault = UserVaultWrapper(app.vault, user)
             # use an invalid symlink path so the posix list would fail if used
             user_vault.write_secret("posix/root_path", "/invalid/root")
@@ -95,7 +94,7 @@ def test_vault_secret_per_app_in_file_source(self):
     def test_upload_file_from_remote_source(self):
         with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
             app = self._app
-            user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
+            user = get_user_by_email(self._app.model.session, self.USER_1_APP_VAULT_ENTRY)
             user_vault = UserVaultWrapper(app.vault, user)
             # use a valid symlink path so the posix list succeeds
             user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
@@ -120,7 +119,3 @@ def test_upload_file_from_remote_source(self):
                 new_dataset = self.dataset_populator.fetch(payload, assert_ok=True).json()["outputs"][0]
                 content = self.dataset_populator.get_history_dataset_content(history_id, dataset=new_dataset)
                 assert content == "I require access to the vault", content
-
-    def _get_user_by_email(self, email):
-        stmt = select(self._app.model.User).filter(self._app.model.User.email == email).limit(1)
-        return self._app.model.session.scalars(stmt).first()