diff --git a/test/integration/objectstore/test_selection_with_resource_parameters.py b/test/integration/objectstore/test_selection_with_resource_parameters.py
index 63a5c188024..19550075b31 100644
--- a/test/integration/objectstore/test_selection_with_resource_parameters.py
+++ b/test/integration/objectstore/test_selection_with_resource_parameters.py
@@ -3,6 +3,8 @@
 import os
 import string
 
+from sqlalchemy import select
+
 from galaxy.model import Dataset
 from galaxy.model.unittest_utils.store_fixtures import (
     deferred_hda_model_store_dict,
@@ -87,7 +89,7 @@ class TestObjectStoreSelectionWithResourceParameterIntegration(BaseObjectStoreIn
 
     def _assert_no_external_filename(self):
         # Should maybe be its own test case ...
-        for external_filename_tuple in self._app.model.session.query(Dataset.external_filename).all():
+        for external_filename_tuple in self._app.model.session.execute(select(Dataset.external_filename)).all():
             assert external_filename_tuple[0] is None
 
     def test_objectstore_selection(self):
@@ -181,5 +183,7 @@ class TestObjectStoreSelectionWithResourceParameterIntegration(BaseObjectStoreIn
 
     @property
     def _latest_dataset(self):
-        latest_dataset = self._app.model.session.query(Dataset).order_by(Dataset.table.c.id.desc()).first()
+        latest_dataset = self._app.model.session.scalars(
+            select(Dataset).order_by(Dataset.table.c.id.desc()).limit(1)
+        ).first()
         return latest_dataset
diff --git a/test/integration/objectstore/test_selection_with_user_preferred_object_store.py b/test/integration/objectstore/test_selection_with_user_preferred_object_store.py
index ecb612fe9c5..13cc0b91267 100644
--- a/test/integration/objectstore/test_selection_with_user_preferred_object_store.py
+++ b/test/integration/objectstore/test_selection_with_user_preferred_object_store.py
@@ -8,6 +8,8 @@ from typing import (
     Optional,
 )
 
+from sqlalchemy import select
+
 from galaxy.model import Dataset
 from galaxy_test.base.populators import WorkflowPopulator
 from galaxy_test.base.workflow_fixtures import (
@@ -482,5 +484,7 @@ class TestObjectStoreSelectionWithPreferredObjectStoresIntegration(BaseObjectSto
 
     @property
     def _latest_dataset(self):
-        latest_dataset = self._app.model.session.query(Dataset).order_by(Dataset.table.c.id.desc()).first()
+        latest_dataset = self._app.model.session.scalars(
+            select(Dataset).order_by(Dataset.table.c.id.desc()).limit(1)
+        ).first()
         return latest_dataset
diff --git a/test/integration/test_celery_tasks.py b/test/integration/test_celery_tasks.py
index 41e93814239..30658dc5b3b 100644
--- a/test/integration/test_celery_tasks.py
+++ b/test/integration/test_celery_tasks.py
@@ -1,6 +1,7 @@
 import tarfile
 
 from celery import shared_task
+from sqlalchemy import select
 
 from galaxy.celery import galaxy_task
 from galaxy.celery.tasks import (
@@ -108,9 +109,5 @@ class TestCeleryTasksIntegration(IntegrationTestCase):
 
     @property
     def _latest_hda(self):
-        latest_hda = (
-            self._app.model.session.query(HistoryDatasetAssociation)
-            .order_by(HistoryDatasetAssociation.table.c.id.desc())
-            .first()
-        )
-        return latest_hda
+        stmt = select(HistoryDatasetAssociation).order_by(HistoryDatasetAssociation.table.c.id.desc()).limit(1)
+        return self._app.model.session.scalars(stmt).first()
diff --git a/test/integration/test_job_files.py b/test/integration/test_job_files.py
index fd933e720ea..0ce62dc98f0 100644
--- a/test/integration/test_job_files.py
+++ b/test/integration/test_job_files.py
@@ -19,6 +19,7 @@ import os
 import tempfile
 
 import requests
+from sqlalchemy import select
 
 from galaxy import model
 from galaxy.model.base import transaction
@@ -50,10 +51,11 @@ class TestJobFilesIntegration(integration_util.IntegrationTestCase):
         if not TestJobFilesIntegration.initialized:
             history_id = self.dataset_populator.new_history()
             sa_session = self.sa_session
-            assert len(sa_session.query(model.HistoryDatasetAssociation).all()) == 0
+            stmt = select(model.HistoryDatasetAssociation)
+            assert len(sa_session.scalars(stmt).all()) == 0
             self.dataset_populator.new_dataset(history_id, content=TEST_INPUT_TEXT, wait=True)
-            assert len(sa_session.query(model.HistoryDatasetAssociation).all()) == 1
-            self.input_hda = sa_session.query(model.HistoryDatasetAssociation).all()[0]
+            assert len(sa_session.scalars(stmt).all()) == 1
+            self.input_hda = sa_session.scalars(stmt).all()[0]
             TestJobFilesIntegration.initialized = True
 
     def test_read_by_state(self):
@@ -119,11 +121,11 @@ class TestJobFilesIntegration(integration_util.IntegrationTestCase):
     def create_static_job_with_state(self, state):
         """Create a job with unknown handler so its state won't change."""
         sa_session = self.sa_session
-        hda = sa_session.query(model.HistoryDatasetAssociation).all()[0]
+        hda = sa_session.scalars(select(model.HistoryDatasetAssociation)).all()[0]
         assert hda
-        history = sa_session.query(model.History).all()[0]
+        history = sa_session.scalars(select(model.History)).all()[0]
         assert history
-        user = sa_session.query(model.User).all()[0]
+        user = sa_session.scalars(select(model.User)).all()[0]
         assert user
         output_hda = model.HistoryDatasetAssociation(history=history, create_dataset=True, flush=False)
         output_hda.hid = 2
diff --git a/test/integration/test_kubernetes_runner.py b/test/integration/test_kubernetes_runner.py
index fbf4f4ed022..62a353693ea 100644
--- a/test/integration/test_kubernetes_runner.py
+++ b/test/integration/test_kubernetes_runner.py
@@ -269,8 +269,8 @@ class TestKubernetesIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJob
             job_dict = running_response["jobs"][0]
 
             app = self._app
-            sa_session = app.model.context.current
-            job = sa_session.query(app.model.Job).get(app.security.decode_id(job_dict["id"]))
+            sa_session = app.model.session
+            job = sa_session.get(app.model.Job, app.security.decode_id(job_dict["id"]))
 
             self._wait_for_external_state(sa_session, job, app.model.Job.states.RUNNING)
             assert not job.finished
@@ -306,8 +306,8 @@ class TestKubernetesIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJob
             job_dict = running_response.json()["jobs"][0]
 
             app = self._app
-            sa_session = app.model.context.current
-            job = sa_session.query(app.model.Job).get(app.security.decode_id(job_dict["id"]))
+            sa_session = app.model.session
+            job = sa_session.get(app.model.Job, app.security.decode_id(job_dict["id"]))
 
             self._wait_for_external_state(sa_session, job, app.model.Job.states.RUNNING)
 
@@ -340,8 +340,8 @@ class TestKubernetesIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJob
         # check that logs are also available in job logs
         app = self._app
         job_id = app.security.decode_id(running_response.json()["jobs"][0]["id"])
-        sa_session = app.model.context
-        job = sa_session.query(app.model.Job).get(job_id)
+        sa_session = app.model.session
+        job = sa_session.get(app.model.Job, job_id)
         self._wait_for_external_state(sa_session=sa_session, job=job, expected=app.model.Job.states.RUNNING)
 
         external_id = job.job_runner_external_id
diff --git a/test/integration/test_local_job_cancellation.py b/test/integration/test_local_job_cancellation.py
index 05ff6a97d3b..888a62b662c 100644
--- a/test/integration/test_local_job_cancellation.py
+++ b/test/integration/test_local_job_cancellation.py
@@ -3,6 +3,7 @@
 import time
 
 import psutil
+from sqlalchemy import select
 
 from galaxy.model.base import transaction
 from galaxy_test.base.populators import DatasetPopulator
@@ -43,13 +44,12 @@ class TestLocalJobCancellation(CancelsJob, integration_util.IntegrationTestCase)
         with self.dataset_populator.test_history() as history_id:
             job_id = self._setup_cat_data_and_sleep(history_id)
             self._wait_for_job_running(job_id)
-            app = self._app
-            sa_session = app.model.context.current
-            Job = app.model.Job
-            job = sa_session.query(Job).filter_by(tool_id="cat_data_and_sleep").order_by(Job.create_time.desc()).first()
+            sa_session = self._app.model.session
+            Job = self._app.model.Job
+            job = self._get_job_by_tool("cat_data_and_sleep")
             # This is how the admin controller code cancels a job
             job.job_stderr = "admin cancelled job"
-            job.set_state(app.model.Job.states.DELETING)
+            job.set_state(Job.states.DELETING)
             sa_session.add(job)
             with transaction(sa_session):
                 sa_session.commit()
@@ -64,17 +64,16 @@ class TestLocalJobCancellation(CancelsJob, integration_util.IntegrationTestCase)
         with self.dataset_populator.test_history() as history_id:
             job_id = self._setup_cat_data_and_sleep(history_id)
 
-            app = self._app
-            sa_session = app.model.context.current
+            sa_session = self._app.model.session
             external_id = None
             state = False
-            Job = app.model.Job
+            Job = self._app.model.Job
 
-            job = sa_session.query(Job).filter_by(tool_id="cat_data_and_sleep").order_by(Job.create_time.desc()).first()
+            job = self._get_job_by_tool("cat_data_and_sleep")
             # Not checking the state here allows the change from queued to running to overwrite
             # the change from queued to deleted_new in the API thread - this is a problem because
             # the job will still run. See issue https://github.com/galaxyproject/galaxy/issues/4960.
-            while external_id is None or state != app.model.Job.states.RUNNING:
+            while external_id is None or state != Job.states.RUNNING:
                 sa_session.refresh(job)
                 assert not job.finished
                 external_id = job.job_runner_external_id
@@ -94,7 +93,7 @@ class TestLocalJobCancellation(CancelsJob, integration_util.IntegrationTestCase)
             for _ in range(100):
                 sa_session.refresh(job)
                 state = job.state
-                if state == app.model.Job.states.DELETED:
+                if state == Job.states.DELETED:
                     break
                 time.sleep(0.1)
 
@@ -106,5 +105,10 @@ class TestLocalJobCancellation(CancelsJob, integration_util.IntegrationTestCase)
                 time.sleep(0.1)
 
             final_state = f"pid exists? {pid_exists}, final db job state {state}"
-            assert state == app.model.Job.states.DELETED, final_state
+            assert state == Job.states.DELETED, final_state
             assert not pid_exists, final_state
+
+    def _get_job_by_tool(self, tool_id):
+        model = self._app.model
+        stmt = select(model.Job).filter_by(tool_id=tool_id).order_by(model.Job.create_time.desc()).limit(1)
+        return model.session.scalars(stmt).first()
diff --git a/test/integration/test_page_revision_json_encoding.py b/test/integration/test_page_revision_json_encoding.py
index 0204298010e..1b7e7a9da2f 100644
--- a/test/integration/test_page_revision_json_encoding.py
+++ b/test/integration/test_page_revision_json_encoding.py
@@ -5,6 +5,8 @@ the security parameter to encode IDs may be changed by admins). Test case also v
 exported API values are encoded though.
 """
 
+from sqlalchemy import select
+
 from galaxy import model
 from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import DatasetPopulator
@@ -26,8 +28,8 @@ class TestPageJsonEncodingIntegration(integration_util.IntegrationTestCase):
         )
         page_response = self._post("pages", request, json=True)
         api_asserts.assert_status_code_is_ok(page_response)
-        sa_session = self._app.model.context
-        page_revision = sa_session.query(model.PageRevision).filter_by(content_format="html").all()[0]
+        sa_session = self._app.model.session
+        page_revision = sa_session.scalars(select(model.PageRevision).filter_by(content_format="html")).all()[0]
         assert '''id="History-1"''' in page_revision.content, page_revision.content
         assert f'''id="History-{history_id}"''' not in page_revision.content, page_revision.content
 
@@ -52,8 +54,8 @@ history_dataset_display(history_dataset_id={})
         )
         page_response = self._post("pages", request, json=True)
         api_asserts.assert_status_code_is_ok(page_response)
-        sa_session = self._app.model.context
-        page_revision = sa_session.query(model.PageRevision).filter_by(content_format="markdown").all()[0]
+        sa_session = self._app.model.session
+        page_revision = sa_session.scalars(select(model.PageRevision).filter_by(content_format="markdown")).all()[0]
         assert (
             """```galaxy
 history_dataset_display(history_dataset_id=1)
diff --git a/test/integration/test_repository_operations.py b/test/integration/test_repository_operations.py
index 4a1523090e0..c5b98098227 100644
--- a/test/integration/test_repository_operations.py
+++ b/test/integration/test_repository_operations.py
@@ -1,6 +1,8 @@
 import os
 from collections import namedtuple
 
+from sqlalchemy import select
+
 from galaxy.model.base import transaction
 from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
@@ -86,7 +88,7 @@ class TestRepositoryInstallIntegrationTestCase(integration_util.IntegrationTestC
         hg_util.update_repository(repository_path, ctx_rev="3")
         # change repo to revision 3 in database
         model = self._app.install_model
-        tsr = model.context.query(model.ToolShedRepository).first()
+        tsr = model.session.scalars(select(model.ToolShedRepository).limit(1)).first()
         assert tsr.name == REPO.name
         assert tsr.changeset_revision == latest_revision
         assert int(tsr.ctx_rev) >= 4
diff --git a/test/integration/test_save_job_id_on_datasets.py b/test/integration/test_save_job_id_on_datasets.py
index 8b59a0a4a52..99685b4ed03 100644
--- a/test/integration/test_save_job_id_on_datasets.py
+++ b/test/integration/test_save_job_id_on_datasets.py
@@ -8,6 +8,7 @@ Here we test the creation of this reference in various contexts:
 """
 
 import pytest
+from sqlalchemy import select
 
 from galaxy import model
 from galaxy_test.driver.driver_util import GalaxyTestDriver
@@ -39,8 +40,8 @@ def test_driver():
 def test_tool_datasets(tool_id, test_driver):
     test_driver.run_tool_test(tool_id)
     session = test_driver.app.model.context.current
-    job = session.query(model.Job).order_by(model.Job.id.desc()).first()
-    datasets = session.query(model.Dataset).filter(model.Dataset.job_id == job.id).all()
+    job = session.scalars(select(model.Job).order_by(model.Job.id.desc()).limit(1)).first()
+    datasets = session.scalars(select(model.Dataset).filter(model.Dataset.job_id == job.id)).all()
 
     if tool_id == "boolean_conditional":
         assert len(datasets) == 1
diff --git a/test/integration/test_user_preferences.py b/test/integration/test_user_preferences.py
index 9ce0f0ed316..84dd304b643 100644
--- a/test/integration/test_user_preferences.py
+++ b/test/integration/test_user_preferences.py
@@ -7,6 +7,7 @@ from requests import (
     get,
     put,
 )
+from sqlalchemy import select
 
 from galaxy_test.driver import integration_util
 
@@ -18,7 +19,8 @@ class TestUserPreferences(integration_util.IntegrationTestCase):
         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)
-        db_user = app.model.context.query(app.model.User).filter(app.model.User.email == user["email"]).first()
+        stmt = select(app.model.User).filter(app.model.User.email == user["email"]).limit(1)
+        db_user = app.model.session.scalars(stmt).first()
 
         # create some initial data
         put(url)
diff --git a/test/integration/test_vault_extra_prefs.py b/test/integration/test_vault_extra_prefs.py
index 0dc67666537..4599d2decbe 100644
--- a/test/integration/test_vault_extra_prefs.py
+++ b/test/integration/test_vault_extra_prefs.py
@@ -9,6 +9,7 @@ from requests import (
     get,
     put,
 )
+from sqlalchemy import select
 
 from galaxy_test.driver import integration_util
 
@@ -28,7 +29,7 @@ class TestExtraUserPreferences(integration_util.IntegrationTestCase, integration
         user = self._setup_user(TEST_USER_EMAIL)
         url = self.__url("information/inputs", user)
         app = cast(Any, self._test_driver.app if self._test_driver else None)
-        db_user = app.model.context.query(app.model.User).filter(app.model.User.email == user["email"]).first()
+        db_user = self._get_dbuser(app, user)
 
         # create some initial data
         put(
@@ -84,7 +85,7 @@ class TestExtraUserPreferences(integration_util.IntegrationTestCase, integration
         user = self._setup_user(TEST_USER_EMAIL)
         url = self.__url("information/inputs", user)
         app = cast(Any, self._test_driver.app if self._test_driver else None)
-        db_user = app.model.context.query(app.model.User).filter(app.model.User.email == user["email"]).first()
+        db_user = self._get_dbuser(app, user)
 
         # write the initial secret value
         put(
@@ -130,3 +131,7 @@ class TestExtraUserPreferences(integration_util.IntegrationTestCase, integration
 
     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()
diff --git a/test/integration/test_vault_file_source.py b/test/integration/test_vault_file_source.py
index 32654d98c3f..ab6295058ff 100644
--- a/test/integration/test_vault_file_source.py
+++ b/test/integration/test_vault_file_source.py
@@ -1,6 +1,8 @@
 import os
 import tempfile
 
+from sqlalchemy import select
+
 from galaxy.security.vault import UserVaultWrapper
 from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import DatasetPopulator
@@ -31,14 +33,11 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
         This file source performs a user vault lookup. The secret stored for the first user is a
         valid path and should succeed, while the second user's stored secret should fail.
         """
+        app = self._app
+
         with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
-            app = self._app
-            user = (
-                app.model.context.query(app.model.User)
-                .filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
-                .first()
-            )
-            user_vault = UserVaultWrapper(self._app.vault, user)
+            user = self._get_user_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])
 
@@ -49,13 +48,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
             print(remote_files)
 
         with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
-            app = self._app
-            user = (
-                app.model.context.query(app.model.User)
-                .filter(app.model.User.email == self.USER_2_APP_VAULT_ENTRY)
-                .first()
-            )
-            user_vault = UserVaultWrapper(self._app.vault, user)
+            user = self._get_user_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")
 
@@ -75,12 +69,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
         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 = (
-                app.model.context.query(app.model.User)
-                .filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
-                .first()
-            )
-            user_vault = UserVaultWrapper(self._app.vault, user)
+            user = self._get_user_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])
 
@@ -91,12 +81,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
             print(remote_files)
 
         with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
-            user = (
-                app.model.context.query(app.model.User)
-                .filter(app.model.User.email == self.USER_2_APP_VAULT_ENTRY)
-                .first()
-            )
-            user_vault = UserVaultWrapper(self._app.vault, user)
+            user = self._get_user_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")
 
@@ -109,12 +95,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
     def test_upload_file_from_remote_source(self):
         with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
             app = self._app
-            user = (
-                app.model.context.query(app.model.User)
-                .filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
-                .first()
-            )
-            user_vault = UserVaultWrapper(self._app.vault, user)
+            user = self._get_user_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])
             data = {"target": "gxfiles://test_user_vault"}
@@ -138,3 +120,7 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
                 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()
diff --git a/test/integration/test_workflow_handler_configuration.py b/test/integration/test_workflow_handler_configuration.py
index c9441459eeb..3e2893a8d81 100644
--- a/test/integration/test_workflow_handler_configuration.py
+++ b/test/integration/test_workflow_handler_configuration.py
@@ -127,8 +127,7 @@ class BaseWorkflowHandlerConfigurationTestCase(integration_util.IntegrationTestC
         # into Galaxy's internal state.
         app = self._app
         history_id = app.security.decode_id(history_id)
-        sa_session = app.model.context.current
-        history = sa_session.query(app.model.History).get(history_id)
+        history = app.model.session.get(app.model.History, history_id)
         workflow_invocations = history.workflow_invocations
         return workflow_invocations
 
diff --git a/test/integration/test_workflow_refactoring.py b/test/integration/test_workflow_refactoring.py
index 47699617411..4048fcc9743 100644
--- a/test/integration/test_workflow_refactoring.py
+++ b/test/integration/test_workflow_refactoring.py
@@ -6,6 +6,8 @@ from typing import (
     List,
 )
 
+from sqlalchemy import select
+
 from galaxy.managers.context import ProvidesAppContext
 from galaxy.managers.workflows import RefactorRequest
 from galaxy.model import (
@@ -767,7 +769,8 @@ steps:
             yield workflow_object
 
     def _refactor(self, actions: List[Dict[str, Any]], stored_workflow=None, dry_run=False, style="ga"):
-        user = self._app.model.session.query(User).order_by(User.id.desc()).limit(1).one()
+        stmt = select(User).order_by(User.id.desc()).limit(1)
+        user = self._app.model.session.execute(stmt).scalar_one()
         mock_trans = MockTrans(self._app, user)
 
         app = self._app
@@ -818,11 +821,13 @@ steps:
         return response
 
     def _model_last_time(self, clazz):
-        obj = self._app.model.session.query(clazz).order_by(clazz.update_time.desc()).limit(1).one()
+        stmt = select(clazz).order_by(clazz.update_time.desc()).limit(1)
+        obj = self._app.model.session.execute(stmt).unique().scalar_one()
         return obj.update_time
 
     def _model_last_id(self, clazz):
-        obj = self._app.model.session.query(clazz).order_by(clazz.id.desc()).limit(1).one_or_none()
+        stmt = select(clazz).order_by(clazz.id.desc()).limit(1)
+        obj = self._app.model.session.execute(stmt).scalar_one_or_none()
         return obj.id if obj else None
 
     @property
@@ -835,7 +840,8 @@ steps:
 
     def _recent_stored_workflow(self, n=1):
         app = self._app
-        return app.model.session.query(StoredWorkflow).order_by(StoredWorkflow.id.desc()).limit(n).all()[-1]
+        stmt = select(StoredWorkflow).order_by(StoredWorkflow.id.desc()).limit(n)
+        return app.model.session.scalars(stmt).unique().all()[-1]
 
     @property
     def _latest_workflow(self):
diff --git a/test/unit/app/managers/test_DatasetManager.py b/test/unit/app/managers/test_DatasetManager.py
index 5ef338accdc..882e4a97f71 100644
--- a/test/unit/app/managers/test_DatasetManager.py
+++ b/test/unit/app/managers/test_DatasetManager.py
@@ -3,6 +3,7 @@
 from unittest import mock
 
 import sqlalchemy
+from sqlalchemy import select
 
 from galaxy import (
     exceptions,
@@ -32,14 +33,14 @@ class TestDatasetManager(BaseTestCase):
         self.log("should be able to create a new Dataset")
         dataset1 = self.dataset_manager.create()
         assert isinstance(dataset1, model.Dataset)
-        assert dataset1 == self.trans.sa_session.query(model.Dataset).get(dataset1.id)
+        assert dataset1 == self.trans.sa_session.get(model.Dataset, dataset1.id)
 
     def test_base(self):
         dataset1 = self.dataset_manager.create()
         dataset2 = self.dataset_manager.create()
 
         self.log("should be able to query")
-        datasets = self.trans.sa_session.query(model.Dataset).all()
+        datasets = self.trans.sa_session.scalars(select(model.Dataset)).all()
         assert self.dataset_manager.list() == datasets
         assert self.dataset_manager.one(filters=(model.Dataset.id == dataset1.id)) == dataset1
         assert self.dataset_manager.by_id(dataset1.id) == dataset1
@@ -187,18 +188,6 @@ class TestDatasetManager(BaseTestCase):
         assert not self.dataset_manager.permissions.access.is_permitted(dataset, None)
 
 
-# class TestDatasetRBACPermissions(BaseTestCase):
-#     def set_up_managers(self):
-#         super().set_up_managers()
-#         self.dataset_manager = DatasetManager(self.app)
-#
-#     def test_manage( self ):
-#         self.log("should be able to create a new Dataset")
-#         dataset1 = self.dataset_manager.create()
-#         assert isinstance(dataset1, model.Dataset)
-#         assert dataset1 == self.app.model.context.query(model.Dataset).get(dataset1.id)
-
-
 # web.url_for doesn't work well in the framework
 def testable_url_for(*a, **k):
     return f"(fake url): {a}, {k}"
diff --git a/test/unit/app/managers/test_HDAManager.py b/test/unit/app/managers/test_HDAManager.py
index 94ebfbb0b4a..38a241fd04a 100644
--- a/test/unit/app/managers/test_HDAManager.py
+++ b/test/unit/app/managers/test_HDAManager.py
@@ -1,6 +1,7 @@
 from unittest import mock
 
 import sqlalchemy
+from sqlalchemy import select
 
 from galaxy import (
     exceptions,
@@ -44,7 +45,7 @@ class TestHDAManager(HDATestCase):
         hda3 = self.hda_manager.create(history=history1, hid=3)
 
         self.log("should be able to query")
-        hdas = self.trans.sa_session.query(hda_model).all()
+        hdas = self.trans.sa_session.scalars(select(hda_model)).all()
         assert self.hda_manager.list() == hdas
         assert self.hda_manager.one(filters=(hda_model.id == hda1.id)) == hda1
         assert self.hda_manager.by_id(hda1.id) == hda1
@@ -70,7 +71,7 @@ class TestHDAManager(HDATestCase):
         self.log("should be able to create a new HDA with a specified history and dataset")
         hda1 = self.hda_manager.create(history=history1, dataset=dataset1)
         assert isinstance(hda1, model.HistoryDatasetAssociation)
-        assert hda1 == self.trans.sa_session.query(model.HistoryDatasetAssociation).get(hda1.id)
+        assert hda1 == self.trans.sa_session.get(model.HistoryDatasetAssociation, hda1.id)
         assert hda1.history == history1
         assert hda1.dataset == dataset1
         assert hda1.hid == 1
@@ -121,7 +122,7 @@ class TestHDAManager(HDATestCase):
         self.log("should be able to copy an HDA")
         hda2 = self.hda_manager.copy(hda1, history=history1)
         assert isinstance(hda2, model.HistoryDatasetAssociation)
-        assert hda2 == self.trans.sa_session.query(model.HistoryDatasetAssociation).get(hda2.id)
+        assert hda2 == self.trans.sa_session.get(model.HistoryDatasetAssociation, hda2.id)
         assert hda2.name == hda1.name
         assert hda2.history == hda1.history
         assert hda2.dataset == hda1.dataset
diff --git a/test/unit/app/managers/test_HistoryManager.py b/test/unit/app/managers/test_HistoryManager.py
index 9957f84c43c..42ed9de4ff7 100644
--- a/test/unit/app/managers/test_HistoryManager.py
+++ b/test/unit/app/managers/test_HistoryManager.py
@@ -4,6 +4,7 @@ import pytest
 import sqlalchemy
 from sqlalchemy import (
     false,
+    select,
     true,
 )
 
@@ -51,14 +52,22 @@ class TestHistoryManager(BaseTestCase):
         assert isinstance(history1, model.History)
         assert history1.name == "history1"
         assert history1.user == user2
-        assert history1 == self.trans.sa_session.query(model.History).get(history1.id)
-        assert history1 == self.trans.sa_session.query(model.History).filter(model.History.name == "history1").one()
-        assert history1 == self.trans.sa_session.query(model.History).filter(model.History.user == user2).one()
+        assert history1 == self.trans.sa_session.get(model.History, history1.id)
+        assert (
+            history1
+            == self.trans.sa_session.execute(
+                select(model.History).filter(model.History.name == "history1")
+            ).scalar_one()
+        )
+        assert (
+            history1
+            == self.trans.sa_session.execute(select(model.History).filter(model.History.user == user2)).scalar_one()
+        )
 
         history2 = self.history_manager.copy(history1, user=user3)
 
         self.log("should be able to query")
-        histories = self.trans.sa_session.query(model.History).all()
+        histories = self.trans.sa_session.scalars(select(model.History)).all()
         assert self.history_manager.one(filters=(model.History.id == history1.id)) == history1
         assert self.history_manager.list() == histories
         assert self.history_manager.by_id(history1.id) == history1
@@ -97,7 +106,7 @@ class TestHistoryManager(BaseTestCase):
         history2 = self.history_manager.copy(history1, user=user3)
         assert isinstance(history2, model.History)
         assert history2.user == user3
-        assert history2 == self.trans.sa_session.query(model.History).get(history2.id)
+        assert history2 == self.trans.sa_session.get(model.History, history2.id)
         assert history2.name == history1.name
         assert history2 != history1
 
diff --git a/test/unit/app/managers/test_UserManager.py b/test/unit/app/managers/test_UserManager.py
index 00152b51ca9..0fa3231148d 100644
--- a/test/unit/app/managers/test_UserManager.py
+++ b/test/unit/app/managers/test_UserManager.py
@@ -6,7 +6,10 @@ Executable directly using: python -m test.unit.managers.test_UserManager
 from datetime import datetime
 from unittest.mock import patch
 
-from sqlalchemy import desc
+from sqlalchemy import (
+    desc,
+    select,
+)
 
 from galaxy import (
     exceptions,
@@ -47,7 +50,7 @@ class TestUserManager(BaseTestCase):
         user3 = self.user_manager.create(**user3_data)
 
         self.log("should be able to query")
-        users = self.trans.sa_session.query(model.User).all()
+        users = self.trans.sa_session.scalars(select(model.User)).all()
         assert self.user_manager.list() == users
 
         assert self.user_manager.by_id(user2.id) == user2
diff --git a/test/unit/app/tools/test_execution.py b/test/unit/app/tools/test_execution.py
index 97b4c4920ee..516a75d24bf 100644
--- a/test/unit/app/tools/test_execution.py
+++ b/test/unit/app/tools/test_execution.py
@@ -4,6 +4,7 @@ from collections import OrderedDict
 from typing import cast
 
 import webob.exc
+from sqlalchemy import select
 
 import galaxy.model
 from galaxy.app_unittest_utils import tools_support
@@ -198,7 +199,7 @@ class MockTrans:
         self.user = None
         self.history._active_datasets_and_roles = [
             hda
-            for hda in self.app.model.context.query(galaxy.model.HistoryDatasetAssociation).all()
+            for hda in self.app.model.session.scalars(select(galaxy.model.HistoryDatasetAssociation)).all()
             if hda.active and hda.history == history
         ]
         self.workflow_building_mode = False
diff --git a/test/unit/app/tools/test_history_imp_exp.py b/test/unit/app/tools/test_history_imp_exp.py
index 32c4cd294d8..3b2891b90fb 100644
--- a/test/unit/app/tools/test_history_imp_exp.py
+++ b/test/unit/app/tools/test_history_imp_exp.py
@@ -5,6 +5,8 @@ import tempfile
 from shutil import rmtree
 from unittest.mock import Mock
 
+from sqlalchemy import select
+
 from galaxy import model
 from galaxy.app_unittest_utils.galaxy_mock import MockApp
 from galaxy.exceptions import MalformedContents
@@ -704,7 +706,8 @@ def test_import_1901_default():
     # There was a deleted dataset so skip to 3
     assert dataset1.hid == 3, dataset1.hid
 
-    jobs = app.model.context.query(model.Job).filter_by(history_id=new_history.id).order_by(model.Job.table.c.id).all()
+    stmt = select(model.Job).filter_by(history_id=new_history.id).order_by(model.Job.id)
+    jobs = app.model.session.scalars(stmt).all()
     assert len(jobs) == 2
     assert jobs[0].tool_id == "upload1"
     assert jobs[1].tool_id == "cat"
diff --git a/test/unit/data/model/test_model_discovery.py b/test/unit/data/model/test_model_discovery.py
index 751fb41ec81..ba32fdaf395 100644
--- a/test/unit/data/model/test_model_discovery.py
+++ b/test/unit/data/model/test_model_discovery.py
@@ -1,6 +1,8 @@
 import os
 from tempfile import mkdtemp
 
+from sqlalchemy import select
+
 from galaxy import model
 from galaxy.model import store
 from galaxy.model.base import transaction
@@ -211,7 +213,7 @@ def test_persist_target_hdca():
 
 
 def _assert_one_library_created(sa_session):
-    all_libraries = sa_session.query(model.Library).all()
+    all_libraries = sa_session.scalars(select(model.Library)).all()
     assert len(all_libraries) == 1, len(all_libraries)
     new_library = all_libraries[0]
     return new_library
diff --git a/test/unit/data/model/test_model_store.py b/test/unit/data/model/test_model_store.py
index bcc7c58cd09..54786a213c1 100644
--- a/test/unit/data/model/test_model_store.py
+++ b/test/unit/data/model/test_model_store.py
@@ -16,6 +16,7 @@ from typing import (
 
 import pytest
 from rocrate.rocrate import ROCrate
+from sqlalchemy import select
 from sqlalchemy.orm.scoping import scoped_session
 
 from galaxy import model
@@ -195,9 +196,9 @@ def test_import_library_from_dict():
     perform_import_from_store_dict(fixture_context, import_dict, import_options=import_options)
 
     sa_session = fixture_context.sa_session
-    all_libraries = sa_session.query(model.Library).all()
+    all_libraries = sa_session.scalars(select(model.Library)).all()
     assert len(all_libraries) == 1, len(all_libraries)
-    all_lddas = sa_session.query(model.LibraryDatasetDatasetAssociation).all()
+    all_lddas = sa_session.scalars(select(model.LibraryDatasetDatasetAssociation)).all()
     assert len(all_lddas) == 1, len(all_lddas)
 
 
@@ -308,9 +309,9 @@ def test_import_export_library():
     )
     import_model_store.perform_import()
 
-    all_libraries = sa_session.query(model.Library).all()
+    all_libraries = sa_session.scalars(select(model.Library)).all()
     assert len(all_libraries) == 2, len(all_libraries)
-    all_lddas = sa_session.query(model.LibraryDatasetDatasetAssociation).all()
+    all_lddas = sa_session.scalars(select(model.LibraryDatasetDatasetAssociation)).all()
     assert len(all_lddas) == 2, len(all_lddas)
 
     new_library = [lib for lib in all_libraries if lib.id != library.id][0]
diff --git a/test/unit/data/test_dataset_materialization.py b/test/unit/data/test_dataset_materialization.py
index ea3dfa5f357..c107f2eb6bc 100644
--- a/test/unit/data/test_dataset_materialization.py
+++ b/test/unit/data/test_dataset_materialization.py
@@ -1,4 +1,5 @@
 from pkg_resources import resource_string
+from sqlalchemy import select
 
 from galaxy.files.unittest_utils import TestPosixConfiguredFileSources
 from galaxy.model import (
@@ -154,7 +155,7 @@ def test_deferred_ldda_basic_attached():
     fixture_context = setup_fixture_context_with_history()
     store_dict = one_ld_library_deferred_model_store_dict()
     perform_import_from_store_dict(fixture_context, store_dict, import_options=import_options)
-    deferred_ldda = fixture_context.sa_session.query(LibraryDatasetDatasetAssociation).all()[0]
+    deferred_ldda = fixture_context.sa_session.scalars(select(LibraryDatasetDatasetAssociation)).all()[0]
     assert deferred_ldda
     assert deferred_ldda.dataset.state == "deferred"
 
diff --git a/test/unit/data/test_galaxy_mapping.py b/test/unit/data/test_galaxy_mapping.py
index b1679332e49..0f54170535f 100644
--- a/test/unit/data/test_galaxy_mapping.py
+++ b/test/unit/data/test_galaxy_mapping.py
@@ -53,10 +53,6 @@ class BaseModelTestCase(TestCase):
         cls.model = mapping.init("/tmp", cls._db_uri(), create_tables=True, object_store=MockObjectStore())
         assert cls.model.engine is not None
 
-    @classmethod
-    def query(cls, type):
-        return cls.model.session.query(type)
-
     @classmethod
     def persist(cls, *args, **kwargs):
         session = cls.session()
@@ -90,7 +86,7 @@ class TestMappings(BaseModelTestCase):
             rating_association = rating_class(u, item, rating)
             self.persist(rating_association)
             self.expunge()
-            stored_rating = self.query(rating_class).all()[0]
+            stored_rating = self.model.session.scalars(select(rating_class)).all()[0]
             assert stored_rating.rating == rating
             assert stored_rating.user.email == user_email
 
@@ -329,12 +325,12 @@ class TestMappings(BaseModelTestCase):
 
         self.persist(u, h1, d1, d2, c1, hc1, dce1, dce2)
 
-        loaded_dataset_collection = (
-            self.query(model.HistoryDatasetCollectionAssociation)
+        stmt = (
+            select(model.HistoryDatasetCollectionAssociation)
             .filter(model.HistoryDatasetCollectionAssociation.name == "HistoryCollectionTest1")
-            .first()
-            .collection
+            .limit(1)
         )
+        loaded_dataset_collection = self.model.session.scalars(stmt).first().collection
         assert len(loaded_dataset_collection.elements) == 2
         assert loaded_dataset_collection.collection_type == "pair"
         assert loaded_dataset_collection["left"] == dce1
@@ -502,11 +498,11 @@ class TestMappings(BaseModelTestCase):
         u.adjust_total_disk_usage(1, None)
         u_id = u.id
         self.expunge()
-        user_reload = self.model.session.query(model.User).get(u_id)
+        user_reload = self.model.session.get(model.User, u_id)
         assert user_reload.disk_usage == 1
 
     def test_basic(self):
-        original_user_count = len(self.model.session.query(model.User).all())
+        original_user_count = len(self.model.session.scalars(select(model.User)).all())
 
         # Make some changes and commit them
         u = model.User(email="james@foo.bar.baz", password="password")
@@ -520,14 +516,14 @@ class TestMappings(BaseModelTestCase):
         self.persist(d1)
 
         # Check
-        users = self.model.session.query(model.User).all()
+        users = self.model.session.scalars(select(model.User)).all()
         assert len(users) == original_user_count + 1
         user = [user for user in users if user.email == "james@foo.bar.baz"][0]
         assert user.email == "james@foo.bar.baz"
         assert user.password == "password"
         assert len(user.histories) == 1
         assert user.histories[0].name == "History 1"
-        hists = self.model.session.query(model.History).all()
+        hists = self.model.session.scalars(select(model.History)).all()
         hist0 = [history for history in hists if history.id == h1.id][0]
         hist1 = [history for history in hists if history.id == h2.id][0]
         assert hist0.name == "History 1"
@@ -541,7 +537,7 @@ class TestMappings(BaseModelTestCase):
         # Do an update and check
         hist1.name = "History 2b"
         self.expunge()
-        hists = self.model.session.query(model.History).all()
+        hists = self.model.session.scalars(select(model.History)).all()
         hist0 = [history for history in hists if history.name == "History 1"][0]
         hist1 = [history for history in hists if history.name == "History 2b"][0]
         assert hist0.name == "History 1"
@@ -560,7 +556,9 @@ class TestMappings(BaseModelTestCase):
         job = model.Job()
         dataset.job = job
         self.persist(job, dataset)
-        loaded_dataset = self.model.session.query(model.Dataset).filter(model.Dataset.id == dataset.id).one()
+        loaded_dataset = self.model.session.execute(
+            select(model.Dataset).filter(model.Dataset.id == dataset.id)
+        ).scalar_one()
         assert loaded_dataset.job_id == job.id
 
     def test_jobs(self):
@@ -571,7 +569,7 @@ class TestMappings(BaseModelTestCase):
 
         self.persist(u, job)
 
-        loaded_job = self.model.session.query(model.Job).filter(model.Job.user == u).first()
+        loaded_job = self.model.session.scalars(select(model.Job).filter(model.Job.user == u).limit(1)).first()
         assert loaded_job.tool_id == "cat1"
 
     def test_job_metrics(self):
@@ -602,7 +600,7 @@ class TestMappings(BaseModelTestCase):
         job.user = u
         self.persist(u, job, task)
 
-        loaded_task = self.model.session.query(model.Task).filter(model.Task.job == job).first()
+        loaded_task = self.model.session.scalars(select(model.Task).filter(model.Task.job == job).limit(1)).first()
         assert loaded_task.prepare_input_files_cmd == "split.sh"
 
     def test_history_contents(self):
@@ -620,9 +618,9 @@ class TestMappings(BaseModelTestCase):
         self.session().flush()
 
         def contents_iter_names(**kwds):
-            history = (
-                self.model.context.query(model.History).filter(model.History.name == "HistoryContentsHistory1").first()
-            )
+            history = self.model.session.scalars(
+                select(model.History).filter(model.History.name == "HistoryContentsHistory1").limit(1)
+            ).first()
             return list(map(lambda hda: hda.name, history.contents_iter(**kwds)))
 
         assert contents_iter_names() == ["1", "2", "3", "4"]
@@ -645,12 +643,8 @@ class TestMappings(BaseModelTestCase):
         h2 = model.History(name="HistoryAuditHistory", user=u)
 
         def get_audit_table_entries(history):
-            return (
-                self.session()
-                .query(model.HistoryAudit.table)
-                .filter(model.HistoryAudit.table.c.history_id == history.id)
-                .all()
-            )
+            stmt = select(model.HistoryAudit.table).filter(model.HistoryAudit.table.c.history_id == history.id)
+            return self.session().execute(stmt).all()
 
         def get_latest_entry(entries):
             # key ensures result is correct if new columns are added
@@ -713,7 +707,7 @@ class TestMappings(BaseModelTestCase):
 
         self.expunge()
         session = self.session()
-        galaxy_model_object = self.query(model.GalaxySession).get(galaxy_session_id)
+        galaxy_model_object = self.model.session.get(model.GalaxySession, galaxy_session_id)
         expected_id = galaxy_model_object.id
 
         # id loaded as part of the object query, could be any non-deferred attribute.
@@ -841,7 +835,7 @@ class TestMappings(BaseModelTestCase):
         history_id = h1.id
         self.expunge()
 
-        loaded_invocation = self.query(model.WorkflowInvocation).get(workflow_invocation.id)
+        loaded_invocation = self.model.session.get(model.WorkflowInvocation, workflow_invocation.id)
         assert loaded_invocation.uuid == invocation_uuid, f"{loaded_invocation.uuid} != {invocation_uuid}"
         assert loaded_invocation
         assert loaded_invocation.history.id == history_id
@@ -859,7 +853,7 @@ class TestMappings(BaseModelTestCase):
 
         assert subworkflow_invocation_assoc.subworkflow_invocation.history.id == history_id
 
-        loaded_workflow = self.query(model.Workflow).get(workflow_id)
+        loaded_workflow = self.model.session.get(model.Workflow, workflow_id)
         assert len(loaded_workflow.steps[0].annotations) == 1
         copied_workflow = loaded_workflow.copy(user=user)
         annotations = copied_workflow.steps[0].annotations
diff --git a/test/unit/data/test_mutable_json_column.py b/test/unit/data/test_mutable_json_column.py
index 3397b292db8..895438e74b4 100644
--- a/test/unit/data/test_mutable_json_column.py
+++ b/test/unit/data/test_mutable_json_column.py
@@ -12,7 +12,7 @@ class TestMutableColumn(BaseModelTestCase):
         with transaction(session):
             session.commit()
         self.model.session.expunge_all()
-        return self.model.session.query(model.DynamicTool).get(item_id)
+        return self.model.session.get(model.DynamicTool, item_id)
 
     def test_metadata_mutable_column(self):
         w = model.DynamicTool()
diff --git a/test/unit/test_galaxy_transactions.py b/test/unit/test_galaxy_transactions.py
index d759d8dafc3..8093df91c0c 100644
--- a/test/unit/test_galaxy_transactions.py
+++ b/test/unit/test_galaxy_transactions.py
@@ -1,4 +1,5 @@
 import pytest
+from sqlalchemy import select
 
 from galaxy import model
 from galaxy.managers import context
@@ -36,26 +37,26 @@ def transaction():
 
 def test_logging_events_off(transaction):
     transaction.log_event("test event 123")
-    assert len(transaction.sa_session.query(model.Event).all()) == 0
+    assert len(transaction.sa_session.scalars(select(model.Event)).all()) == 0
 
 
 def test_logging_events_on(transaction):
     transaction.app.config.log_events = True
     transaction.log_event("test event 123")
-    events = transaction.sa_session.query(model.Event).all()
+    events = transaction.sa_session.scalars(select(model.Event)).all()
     assert len(events) == 1
     assert events[0].message == "test event 123"
 
 
 def test_logging_actions_off(transaction):
     transaction.log_action("test action 123")
-    assert len(transaction.sa_session.query(model.Event).all()) == 0
+    assert len(transaction.sa_session.scalars(select(model.Event)).all()) == 0
 
 
 def test_logging_actions_on(transaction):
     transaction.app.config.log_actions = True
     transaction.log_action(None, "test action 123", context="the context", params=dict(foo="bar"))
-    actions = transaction.sa_session.query(model.UserAction).all()
+    actions = transaction.sa_session.scalars(select(model.UserAction)).all()
     assert len(actions) == 1
     assert actions[0].action == "test action 123"
 
@@ -69,7 +70,7 @@ def test_expunge_all(transaction):
     with db_transaction(session):
         session.commit()
 
-    assert transaction.sa_session.query(model.User).first().password == "bar2"
+    assert transaction.sa_session.scalars(select(model.User).limit(1)).first().password == "bar2"
 
     transaction.sa_session.expunge_all()
 
@@ -79,4 +80,4 @@ def test_expunge_all(transaction):
         session.commit()
 
     # Password unchange because not attached to session/context.
-    assert transaction.sa_session.query(model.User).first().password == "bar2"
+    assert transaction.sa_session.scalars(select(model.User).limit(1)).first().password == "bar2"
diff --git a/test/unit/workflows/test_run_parameters.py b/test/unit/workflows/test_run_parameters.py
index 0c25c1625df..76eae895574 100644
--- a/test/unit/workflows/test_run_parameters.py
+++ b/test/unit/workflows/test_run_parameters.py
@@ -124,4 +124,4 @@ def __workflow_fixure(trans):
     workflow_id = workflow.id
     trans.app.model.context.expunge_all()
 
-    return trans.app.model.context.query(model.Workflow).get(workflow_id)
+    return trans.app.model.session.get(model.Workflow, workflow_id)