diff --git a/lib/galaxy/config/__init__.py b/lib/galaxy/config/__init__.py
index 7ca7a1f15a7..fcc8cfcf7ed 100644
--- a/lib/galaxy/config/__init__.py
+++ b/lib/galaxy/config/__init__.py
@@ -135,6 +135,7 @@ LOGGING_CONFIG_DEFAULT: Dict[str, Any] = {
 """Default value for logging configuration, passed to :func:`logging.config.dictConfig`"""
 
 VERSION_JSON_FILE = "version.json"
+DISABLED_FLAG = "disabled"  # Used to mark a config option as disabled
 
 
 def configure_logging(config, facts=None):
@@ -1276,6 +1277,15 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin):
                     f"Config option '{key}' is deprecated and will be removed in a future release.  Please consult the latest version of the sample configuration file."
                 )
 
+    def is_fetch_with_celery_enabled(self):
+        celery_enabled = self.enable_celery_tasks
+        fetch_disabled = (
+            self.celery_conf
+            and self.celery_conf["task_routes"]
+            and self.celery_conf["task_routes"]["galaxy.fetch_data"] == DISABLED_FLAG
+        )
+        return celery_enabled and not fetch_disabled
+
     @staticmethod
     def _parse_allowed_origin_hostnames(allowed_origin_hostnames):
         """
diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py
index a668588e71c..62a9f727439 100644
--- a/lib/galaxy/tools/execute.py
+++ b/lib/galaxy/tools/execute.py
@@ -180,7 +180,7 @@ def execute(
     tool_id = tool.id
     for job2 in execution_tracker.successful_jobs:
         # Put the job in the queue if tracking in memory
-        if tool_id == "__DATA_FETCH__" and tool.app.config.enable_celery_tasks:
+        if tool_id == "__DATA_FETCH__" and tool.app.config.is_fetch_with_celery_enabled:
             job_id = job2.id
             from galaxy.celery.tasks import (
                 fetch_data,
diff --git a/test/unit/config/test_config_values.py b/test/unit/config/test_config_values.py
index 43f34e13bea..699bbe8dceb 100644
--- a/test/unit/config/test_config_values.py
+++ b/test/unit/config/test_config_values.py
@@ -6,7 +6,7 @@ from galaxy import config
 from galaxy.util.properties import running_from_source
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture()
 def appconfig():
     return config.GalaxyAppConfiguration(override_tempdir=False)
 
@@ -42,3 +42,34 @@ def test_base_config_if_running_not_from_source(monkeypatch):
     assert appconfig.config_dir == os.getcwd()
     assert appconfig.data_dir == os.path.join(appconfig.config_dir, "data")
     assert appconfig.managed_config_dir == os.path.join(appconfig.data_dir, "config")
+
+
+class TestIsFetchWithCeleryEnabled:
+    def test_disabled_if_celery_disabled(self, appconfig):
+        appconfig.enable_celery_tasks = False
+        assert not appconfig.is_fetch_with_celery_enabled()
+
+    def test_enabled_if_no_celeryconf(self, appconfig):
+        appconfig.enable_celery_tasks = True
+        appconfig.enable_celery_conf = None
+        assert appconfig.is_fetch_with_celery_enabled()
+
+    def test_enabled_if_no_task_routes(self, appconfig):
+        appconfig.enable_celery_tasks = True
+        appconfig.celery_conf["task_routes"] = None
+        assert appconfig.is_fetch_with_celery_enabled()
+
+    def test_enabled_if_no_route(self, appconfig):
+        appconfig.enable_celery_tasks = True
+        appconfig.celery_conf["task_routes"]["galaxy.fetch_data"] = None
+        assert appconfig.is_fetch_with_celery_enabled()
+
+    def test_enabled_if_has_route(self, appconfig):
+        appconfig.enable_celery_tasks = True
+        appconfig.celery_conf["task_routes"]["galaxy.fetch_data"] = "my_route"
+        assert appconfig.is_fetch_with_celery_enabled()
+
+    def test_disabled_if_disabled_flag(self, appconfig):
+        appconfig.enable_celery_tasks = True
+        appconfig.celery_conf["task_routes"]["galaxy.fetch_data"] = config.DISABLED_FLAG
+        assert not appconfig.is_fetch_with_celery_enabled()