Use UserRepo for get_by_username and get_by_email

lib/galaxy/managers/users.py

@@ -37,6 +37,7 @@
 )
 from galaxy.model import UserQuotaUsage
 from galaxy.model.base import transaction
+from galaxy.model.repositories.user import UserRepository
 from galaxy.security.validate_user_input import (
     VALID_EMAIL_RE,
     validate_email,
@@ -351,10 +352,11 @@ def error_if_anonymous(self, user, msg="Log-in required", **kwargs):
 
     def get_user_by_identity(self, identity):
         """Get user by username or email."""
+        user_repo = UserRepository(self.session())
         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 = user_repo.get_by_email(identity)
             if not user:
                 # Try a case-insensitive match on the email
                 user = (
@@ -364,7 +366,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 = user_repo.get_by_username(identity)
         return user
 
     # ---- current
@@ -527,7 +529,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 = UserRepository(trans.sa_session).get_by_email(email)
         activation_token = user.activation_token
         if activation_token is None:
             activation_token = util.hash_util.new_secure_hash_v2(str(random.getrandbits(256)))
@@ -571,9 +573,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 = UserRepository(trans.sa_session).get_by_email(email)
         if not reset_user and email != email.lower():
             reset_user = (
                 trans.sa_session.query(self.app.model.User)
@@ -617,12 +617,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 = UserRepository(self.session()).get_by_email(remote_user_email)
         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

lib/galaxy/visualization/genomes.py

@@ -6,13 +6,13 @@
 from typing import Dict
 
 from bx.seq.twobit import TwoBitFile
-from sqlalchemy import select
 
 from galaxy.exceptions import (
     ObjectNotFound,
     ReferenceDataError,
 )
 from galaxy.model.repositories.hda import HistoryDatasetAssociationRepository as hda_repo
+from galaxy.model.repositories.user import UserRepository
 from galaxy.structured_app import StructuredApp
 from galaxy.util.bunch import Bunch
 
@@ -293,7 +293,7 @@ def chroms(self, trans, dbkey=None, num=None, chrom=None, low=None):
         # If there is no dbkey owner, default to current user.
         dbkey_owner, dbkey = decode_dbkey(dbkey)
         if dbkey_owner:
-            dbkey_user = self._get_dbkey_user(trans, dbkey_owner)
+            dbkey_user = UserRepository(trans.sa_session).get_by_username(dbkey_owner)
         else:
             dbkey_user = trans.user
 
@@ -370,7 +370,7 @@ def reference(self, trans, dbkey, chrom, low, high):
         # If there is no dbkey owner, default to current user.
         dbkey_owner, dbkey = decode_dbkey(dbkey)
         if dbkey_owner:
-            dbkey_user = self._get_dbkey_user(trans, dbkey_owner)
+            dbkey_user = UserRepository(trans.sa_session).get_by_username(dbkey_owner)
         else:
             dbkey_user = trans.user
 
@@ -405,7 +405,3 @@ def _get_reference_data(twobit_file_name, chrom, low, high):
             if chrom in twobit:
                 seq_data = twobit[chrom].get(int(low), int(high))
                 return GenomeRegion(chrom=chrom, start=low, end=high, sequence=seq_data)
-
-    def _get_dbkey_user(self, trans, dbkey_owner):
-        stmt = select(trans.app.model.User).filter_by(username=dbkey_owner).limit(1)
-        return trans.sa_session.scalars(stmt).first()

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

@@ -18,6 +18,7 @@
 )
 from galaxy.exceptions import Conflict
 from galaxy.managers import users
+from galaxy.model.repositories.user import UserRepository
 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 = UserRepository(trans.sa_session).get_by_email(email)
             if not user:
                 # Probably wrong email address
                 return trans.show_error_message(

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

@@ -18,14 +18,14 @@
     and_,
     not_,
     or_,
-    select,
 )
 
 from galaxy import (
     model,
     util,
 )
 from galaxy.model.repositories.job import JobRepository
+from galaxy.model.repositories.user import UserRepository
 from galaxy.web.legacy_framework import grids
 from galaxy.webapps.base.controller import (
     BaseUIController,
@@ -1306,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 = UserRepository(trans.sa_session).get_by_email(monitor_email)
     if monitor_row is not None:
         monitor_user_id = monitor_row[0]
     return monitor_user_id

lib/tool_shed/webapp/controllers/repository.py

@@ -54,7 +54,6 @@
 from tool_shed.util.web_util import escape
 from tool_shed.utility_containers import ToolShedUtilityContainerManager
 from tool_shed.webapp.framework.decorators import require_login
-from tool_shed.webapp.model import User
 from tool_shed.webapp.model.repositories import UserRepository
 from tool_shed.webapp.util import ratings_util
 

lib/tool_shed/webapp/model/repositories.py

@@ -1,9 +1,9 @@
 from sqlalchemy import select
+
 from tool_shed.webapp.model import User
 
 
 class UserRepository:
-
     def get_by_username(self, session, username: str):
         stmt = select(User).filter(User.username == username).limit(1)
         return session.scalars(stmt).first()

test/integration/test_user_preferences.py

@@ -7,8 +7,8 @@
     get,
     put,
 )
-from sqlalchemy import select
 
+from galaxy.model.repositories.user import UserRepository
 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 = UserRepository(app.model.session).get_by_email(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.model.repositories.user import UserRepository
 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 UserRepository(app.model.session).get_by_email(user["email"])

test/integration/test_vault_file_source.py

@@ -1,8 +1,7 @@
 import os
 import tempfile
 
-from sqlalchemy import select
-
+from galaxy.model.repositories.user import UserRepository
 from galaxy.security.vault import UserVaultWrapper
 from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import DatasetPopulator
@@ -27,6 +26,7 @@ def handle_galaxy_config_kwds(cls, config):
     def setUp(self):
         super().setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.user_repo = UserRepository(self._app.model.session)
 
     def test_vault_secret_per_user_in_file_source(self):
         """
@@ -36,7 +36,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 = self.user_repo.get_by_email(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 +48,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 = self.user_repo.get_by_email(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 +69,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 = self.user_repo.get_by_email(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 +81,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 = self.user_repo.get_by_email(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 +95,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 = self.user_repo.get_by_email(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 +120,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()