diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index 4b2126e491c2a..a70b860d2ee84 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -21,10 +21,11 @@
 import dataclasses
 import logging
 import re
+from collections import defaultdict
 from collections.abc import Hashable
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
-from typing import Any, Callable, cast
+from typing import Any, Callable, cast, Optional, Union
 
 import dateutil.parser
 import numpy as np
@@ -69,7 +70,7 @@
 from sqlalchemy.sql.expression import Label, TextAsFrom
 from sqlalchemy.sql.selectable import Alias, TableClause
 
-from superset import app, db, security_manager
+from superset import app, db, is_feature_enabled, security_manager
 from superset.commands.dataset.exceptions import DatasetNotFoundError
 from superset.common.db_query_status import QueryStatus
 from superset.connectors.sqla.utils import (
@@ -712,6 +713,56 @@ def get_datasource_by_name(
     ) -> BaseDatasource | None:
         raise NotImplementedError()
 
+    def get_template_processor(self, **kwargs: Any) -> BaseTemplateProcessor:
+        raise NotImplementedError()
+
+    def text(self, clause: str) -> TextClause:
+        raise NotImplementedError()
+
+    def get_sqla_row_level_filters(
+        self,
+        template_processor: Optional[BaseTemplateProcessor] = None,
+    ) -> list[TextClause]:
+        """
+        Return the appropriate row level security filters for this table and the
+        current user. A custom username can be passed when the user is not present in the
+        Flask global namespace.
+
+        :param template_processor: The template processor to apply to the filters.
+        :returns: A list of SQL clauses to be ANDed together.
+        """
+        template_processor = template_processor or self.get_template_processor()
+
+        all_filters: list[TextClause] = []
+        filter_groups: dict[Union[int, str], list[TextClause]] = defaultdict(list)
+        try:
+            for filter_ in security_manager.get_rls_filters(self):
+                clause = self.text(
+                    f"({template_processor.process_template(filter_.clause)})"
+                )
+                if filter_.group_key:
+                    filter_groups[filter_.group_key].append(clause)
+                else:
+                    all_filters.append(clause)
+
+            if is_feature_enabled("EMBEDDED_SUPERSET"):
+                for rule in security_manager.get_guest_rls_filters(self):
+                    clause = self.text(
+                        f"({template_processor.process_template(rule['clause'])})"
+                    )
+                    all_filters.append(clause)
+
+            grouped_filters = [or_(*clauses) for clauses in filter_groups.values()]
+            all_filters.extend(grouped_filters)
+            return all_filters
+        except TemplateError as ex:
+            raise QueryObjectValidationError(
+                _(
+                    "Error in jinja expression in RLS filters: %(msg)s",
+                    msg=ex.message,
+                )
+            ) from ex
+
 
 class AnnotationDatasource(BaseDatasource):
     """Dummy object so we can query annotations using 'Viz' objects just like
diff --git a/superset/databases/api.py b/superset/databases/api.py
index 695ea028b476d..eb611837bc9b1 100644
--- a/superset/databases/api.py
+++ b/superset/databases/api.py
@@ -2274,6 +2274,6 @@ def schemas_access_for_file_upload(self, pk: int) -> Response:
         # otherwise the database should have been filtered out
         # in CsvToDatabaseForm
         schemas_allowed_processed = security_manager.get_schemas_accessible_by_user(
-            database, schemas_allowed, True
+            database, database.get_default_catalog(), schemas_allowed, True
         )
         return self.response(200, schemas=schemas_allowed_processed)
diff --git a/superset/extensions/__init__.py b/superset/extensions/__init__.py
index 65ba7eebc8e0d..cc2106b01224c 100644
--- a/superset/extensions/__init__.py
+++ b/superset/extensions/__init__.py
@@ -31,6 +31,7 @@
 from superset.async_events.async_query_manager_factory import AsyncQueryManagerFactory
 from superset.extensions.ssh import SSHManagerFactory
 from superset.extensions.stats_logger import BaseStatsLoggerManager
+from superset.security.manager import SupersetSecurityManager
 from superset.utils.cache_manager import CacheManager
 from superset.utils.encrypt import EncryptedFieldFactory
 from superset.utils.feature_flag_manager import FeatureFlagManager
@@ -84,9 +85,9 @@ def get_files(bundle: str, asset_type: str = "js") -> list[str]:
         return {
             "js_manifest": lambda bundle: get_files(bundle, "js"),
             "css_manifest": lambda bundle: get_files(bundle, "css"),
-            "assets_prefix": self.app.config["STATIC_ASSETS_PREFIX"]
-            if self.app
-            else "",
+            "assets_prefix": (
+                self.app.config["STATIC_ASSETS_PREFIX"] if self.app else ""
+            ),
         }
 
     def parse_manifest_json(self) -> None:
@@ -132,7 +133,7 @@ def init_app(self, app: Flask) -> None:
 migrate = Migrate()
 profiling = ProfilingExtension()
 results_backend_manager = ResultsBackendManager()
-security_manager = LocalProxy(lambda: appbuilder.sm)
+security_manager: SupersetSecurityManager = LocalProxy(lambda: appbuilder.sm)
 ssh_manager_factory = SSHManagerFactory()
 stats_logger_manager = BaseStatsLoggerManager()
 talisman = Talisman()
diff --git a/superset/models/helpers.py b/superset/models/helpers.py
index 1c6ad2f5d3bb0..295ecea70ea45 100644
--- a/superset/models/helpers.py
+++ b/superset/models/helpers.py
@@ -22,7 +22,6 @@
 import logging
 import re
 import uuid
-from collections import defaultdict
 from collections.abc import Hashable
 from datetime import datetime, timedelta
 from typing import Any, cast, NamedTuple, Optional, TYPE_CHECKING, Union
@@ -52,7 +51,7 @@
 from sqlalchemy.sql.selectable import Alias, TableClause
 from sqlalchemy_utils import UUIDType
 
-from superset import app, db, is_feature_enabled, security_manager
+from superset import app, db, is_feature_enabled
 from superset.advanced_data_type.types import AdvancedDataTypeResponse
 from superset.common.db_query_status import QueryStatus
 from superset.common.utils.time_range_utils import get_since_until_from_time_range
@@ -806,47 +805,12 @@ def get_fetch_values_predicate(
 
     def get_sqla_row_level_filters(
         self,
-        template_processor: Optional[BaseTemplateProcessor] = None,
+        template_processor: Optional[BaseTemplateProcessor] = None,  # pylint: disable=unused-argument
     ) -> list[TextClause]:
-        """
-        Return the appropriate row level security filters for this table and the
-        current user. A custom username can be passed when the user is not present in the
-        Flask global namespace.
-
-        :param template_processor: The template processor to apply to the filters.
-        :returns: A list of SQL clauses to be ANDed together.
-        """
-        template_processor = template_processor or self.get_template_processor()
-
-        all_filters: list[TextClause] = []
-        filter_groups: dict[Union[int, str], list[TextClause]] = defaultdict(list)
-        try:
-            for filter_ in security_manager.get_rls_filters(self):
-                clause = self.text(
-                    f"({template_processor.process_template(filter_.clause)})"
-                )
-                if filter_.group_key:
-                    filter_groups[filter_.group_key].append(clause)
-                else:
-                    all_filters.append(clause)
-
-            if is_feature_enabled("EMBEDDED_SUPERSET"):
-                for rule in security_manager.get_guest_rls_filters(self):
-                    clause = self.text(
-                        f"({template_processor.process_template(rule['clause'])})"
-                    )
-                    all_filters.append(clause)
-
-            grouped_filters = [or_(*clauses) for clauses in filter_groups.values()]
-            all_filters.extend(grouped_filters)
-            return all_filters
-        except TemplateError as ex:
-            raise QueryObjectValidationError(
-                _(
-                    "Error in jinja expression in RLS filters: %(msg)s",
-                    msg=ex.message,
-                )
-            ) from ex
+        # TODO: We should refactor this mixin and remove this method
+        # as it exists in the BaseDatasource and is not applicable
+        # for datasources of type query
+        return []
 
     def _process_sql_expression(
         self,
diff --git a/superset/security/manager.py b/superset/security/manager.py
index 90f1199d4b9b0..57a6321ccc784 100644
--- a/superset/security/manager.py
+++ b/superset/security/manager.py
@@ -2646,9 +2646,7 @@ def has_guest_access(self, dashboard: "Dashboard") -> bool:
             return False
 
         dashboards = [
-            r
-            for r in user.resources
-            if r["type"] == GuestTokenResourceType.DASHBOARD.value
+            r for r in user.resources if r["type"] == GuestTokenResourceType.DASHBOARD
         ]
 
         # TODO (embedded): remove this check once uuids are rolled out
diff --git a/tests/integration_tests/security/guest_token_security_tests.py b/tests/integration_tests/security/guest_token_security_tests.py
index f4a515bb75c27..b8bab0390949f 100644
--- a/tests/integration_tests/security/guest_token_security_tests.py
+++ b/tests/integration_tests/security/guest_token_security_tests.py
@@ -113,15 +113,29 @@ def setUp(self) -> None:
         self.authorized_guest = security_manager.get_guest_user_from_token(
             {
                 "user": {},
-                "resources": [{"type": "dashboard", "id": str(self.embedded.uuid)}],
+                "resources": [
+                    {
+                        "type": GuestTokenResourceType.DASHBOARD,
+                        "id": str(self.embedded.uuid),
+                    }
+                ],
+                "iat": 10,
+                "exp": 20,
+                "rls_rules": [],
             }
         )
         self.unauthorized_guest = security_manager.get_guest_user_from_token(
             {
                 "user": {},
                 "resources": [
-                    {"type": "dashboard", "id": "06383667-3e02-4e5e-843f-44e9c5896b6c"}
+                    {
+                        "type": GuestTokenResourceType.DASHBOARD,
+                        "id": "06383667-3e02-4e5e-843f-44e9c5896b6c",
+                    }
                 ],
+                "iat": 10,
+                "exp": 20,
+                "rls_rules": [],
             }
         )
 
@@ -247,15 +261,29 @@ def setUp(self) -> None:
         self.authorized_guest = security_manager.get_guest_user_from_token(
             {
                 "user": {},
-                "resources": [{"type": "dashboard", "id": str(self.embedded.uuid)}],
+                "resources": [
+                    {
+                        "type": GuestTokenResourceType.DASHBOARD,
+                        "id": str(self.embedded.uuid),
+                    }
+                ],
+                "iat": 10,
+                "exp": 20,
+                "rls_rules": [],
             }
         )
         self.unauthorized_guest = security_manager.get_guest_user_from_token(
             {
                 "user": {},
                 "resources": [
-                    {"type": "dashboard", "id": "06383667-3e02-4e5e-843f-44e9c5896b6c"}
+                    {
+                        "type": GuestTokenResourceType.DASHBOARD,
+                        "id": "06383667-3e02-4e5e-843f-44e9c5896b6c",
+                    }
                 ],
+                "iat": 10,
+                "exp": 20,
+                "rls_rules": [],
             }
         )
         self.chart = self.get_slice("Girls")
diff --git a/tests/integration_tests/security/row_level_security_tests.py b/tests/integration_tests/security/row_level_security_tests.py
index 71bb1484e0330..ea744bd9f8689 100644
--- a/tests/integration_tests/security/row_level_security_tests.py
+++ b/tests/integration_tests/security/row_level_security_tests.py
@@ -646,8 +646,15 @@ def guest_user_with_rls(self, rules: Optional[list[Any]] = None) -> GuestUser:
         return security_manager.get_guest_user_from_token(
             {
                 "user": {},
-                "resources": [{"type": GuestTokenResourceType.DASHBOARD.value}],
+                "resources": [
+                    {
+                        "type": GuestTokenResourceType.DASHBOARD,
+                        "id": "06383667-3e02-4e5e-843f-44e9c5896b6c",
+                    }
+                ],
                 "rls_rules": rules,
+                "iat": 10,
+                "exp": 20,
             }
         )
 
diff --git a/tests/unit_tests/charts/commands/importers/v1/import_test.py b/tests/unit_tests/charts/commands/importers/v1/import_test.py
index ddeb2c7431dbf..8284c8565d04b 100644
--- a/tests/unit_tests/charts/commands/importers/v1/import_test.py
+++ b/tests/unit_tests/charts/commands/importers/v1/import_test.py
@@ -76,7 +76,9 @@ def test_import_chart(mocker: MockerFixture, session_with_schema: Session) -> No
     Test importing a chart.
     """
 
-    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
 
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
@@ -89,7 +91,7 @@ def test_import_chart(mocker: MockerFixture, session_with_schema: Session) -> No
     assert chart.external_url is None
 
     # Assert that the can write to chart was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+    mock_can_access.assert_called_once_with("can_write", "Chart")
 
 
 def test_import_chart_managed_externally(
@@ -98,7 +100,9 @@ def test_import_chart_managed_externally(
     """
     Test importing a chart that is managed externally.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
 
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
@@ -111,7 +115,7 @@ def test_import_chart_managed_externally(
     assert chart.external_url == "https://example.org/my_chart"
 
     # Assert that the can write to chart was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+    mock_can_access.assert_called_once_with("can_write", "Chart")
 
 
 def test_import_chart_without_permission(
@@ -121,7 +125,9 @@ def test_import_chart_without_permission(
     """
     Test importing a chart when a user doesn't have permissions to create.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=False)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=False
+    )
 
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
@@ -134,7 +140,7 @@ def test_import_chart_without_permission(
         == "Chart doesn't exist and user doesn't have permission to create charts"
     )
     # Assert that the can write to chart was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+    mock_can_access.assert_called_once_with("can_write", "Chart")
 
 
 def test_filter_chart_annotations(session: Session) -> None:
@@ -162,8 +168,12 @@ def test_import_existing_chart_without_permission(
     """
     Test importing a chart when a user doesn't have permissions to modify.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
-    mocker.patch.object(security_manager, "can_access_chart", return_value=False)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
+    mock_can_access_chart = mocker.patch.object(
+        security_manager, "can_access_chart", return_value=False
+    )
 
     slice = (
         session_with_data.query(Slice)
@@ -180,8 +190,8 @@ def test_import_existing_chart_without_permission(
         )
 
     # Assert that the can write to chart was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Chart")
-    security_manager.can_access_chart.assert_called_once_with(slice)
+    mock_can_access.assert_called_once_with("can_write", "Chart")
+    mock_can_access_chart.assert_called_once_with(slice)
 
 
 def test_import_existing_chart_with_permission(
@@ -191,8 +201,12 @@ def test_import_existing_chart_with_permission(
     """
     Test importing a chart that exists when a user has access permission to that chart.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
-    mocker.patch.object(security_manager, "can_access_chart", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
+    mock_can_access_chart = mocker.patch.object(
+        security_manager, "can_access_chart", return_value=True
+    )
 
     admin = User(
         first_name="Alice",
@@ -215,5 +229,5 @@ def test_import_existing_chart_with_permission(
     with override_user(admin):
         import_chart(config, overwrite=True)
     # Assert that the can write to chart was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Chart")
-    security_manager.can_access_chart.assert_called_once_with(slice)
+    mock_can_access.assert_called_once_with("can_write", "Chart")
+    mock_can_access_chart.assert_called_once_with(slice)
diff --git a/tests/unit_tests/dashboards/commands/importers/v1/import_test.py b/tests/unit_tests/dashboards/commands/importers/v1/import_test.py
index 03f4e9fd3079c..c311f1b3906c0 100644
--- a/tests/unit_tests/dashboards/commands/importers/v1/import_test.py
+++ b/tests/unit_tests/dashboards/commands/importers/v1/import_test.py
@@ -65,7 +65,9 @@ def test_import_dashboard(mocker: MockerFixture, session_with_schema: Session) -
     """
     Test importing a dashboard.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
 
     dashboard = import_dashboard(dashboard_config)
     assert dashboard.dashboard_title == "Test dash"
@@ -73,7 +75,7 @@ def test_import_dashboard(mocker: MockerFixture, session_with_schema: Session) -
     assert dashboard.is_managed_externally is False
     assert dashboard.external_url is None
     # Assert that the can write to dashboard was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+    mock_can_access.assert_called_once_with("can_write", "Dashboard")
 
 
 def test_import_dashboard_managed_externally(
@@ -83,7 +85,9 @@ def test_import_dashboard_managed_externally(
     """
     Test importing a dashboard that is managed externally.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
 
     config = copy.deepcopy(dashboard_config)
     config["is_managed_externally"] = True
@@ -93,7 +97,7 @@ def test_import_dashboard_managed_externally(
     assert dashboard.external_url == "https://example.org/my_dashboard"
 
     # Assert that the can write to dashboard was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+    mock_can_access.assert_called_once_with("can_write", "Dashboard")
 
 
 def test_import_dashboard_without_permission(
@@ -103,7 +107,9 @@ def test_import_dashboard_without_permission(
     """
     Test importing a dashboard when a user doesn't have permissions to create.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=False)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=False
+    )
 
     with pytest.raises(ImportFailedError) as excinfo:
         import_dashboard(dashboard_config)
@@ -113,7 +119,7 @@ def test_import_dashboard_without_permission(
     )
 
     # Assert that the can write to dashboard was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+    mock_can_access.assert_called_once_with("can_write", "Dashboard")
 
 
 def test_import_existing_dashboard_without_permission(
@@ -123,8 +129,12 @@ def test_import_existing_dashboard_without_permission(
     """
     Test importing a dashboard when a user doesn't have permissions to create.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
-    mocker.patch.object(security_manager, "can_access_dashboard", return_value=False)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
+    mock_can_access_dashboard = mocker.patch.object(
+        security_manager, "can_access_dashboard", return_value=False
+    )
 
     dashboard = (
         session_with_data.query(Dashboard)
@@ -141,8 +151,8 @@ def test_import_existing_dashboard_without_permission(
         )
 
     # Assert that the can write to dashboard was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
-    security_manager.can_access_dashboard.assert_called_once_with(dashboard)
+    mock_can_access.assert_called_once_with("can_write", "Dashboard")
+    mock_can_access_dashboard.assert_called_once_with(dashboard)
 
 
 def test_import_existing_dashboard_with_permission(
@@ -152,8 +162,12 @@ def test_import_existing_dashboard_with_permission(
     """
     Test importing a dashboard that exists when a user has access permission to that dashboard.
     """
-    mocker.patch.object(security_manager, "can_access", return_value=True)
-    mocker.patch.object(security_manager, "can_access_dashboard", return_value=True)
+    mock_can_access = mocker.patch.object(
+        security_manager, "can_access", return_value=True
+    )
+    mock_can_access_dashboard = mocker.patch.object(
+        security_manager, "can_access_dashboard", return_value=True
+    )
 
     admin = User(
         first_name="Alice",
@@ -173,5 +187,5 @@ def test_import_existing_dashboard_with_permission(
         import_dashboard(dashboard_config, overwrite=True)
 
     # Assert that the can write to dashboard was checked
-    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
-    security_manager.can_access_dashboard.assert_called_once_with(dashboard)
+    mock_can_access.assert_called_once_with("can_write", "Dashboard")
+    mock_can_access_dashboard.assert_called_once_with(dashboard)
diff --git a/tests/unit_tests/jinja_context_test.py b/tests/unit_tests/jinja_context_test.py
index b2404fba76839..a1ceaa08b2c0e 100644
--- a/tests/unit_tests/jinja_context_test.py
+++ b/tests/unit_tests/jinja_context_test.py
@@ -418,10 +418,6 @@ def test_dataset_macro(mocker: MockerFixture) -> None:
         "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
         return_value=[],
     )
-    mocker.patch(
-        "superset.models.helpers.security_manager.get_rls_filters",
-        return_value=[],
-    )
 
     columns = [
         TableColumn(column_name="ds", is_dttm=1, type="TIMESTAMP"),
@@ -470,10 +466,6 @@ def test_dataset_macro(mocker: MockerFixture) -> None:
         "superset.connectors.sqla.models.security_manager.get_guest_rls_filters",
         return_value=[],
     )
-    mocker.patch(
-        "superset.models.helpers.security_manager.get_guest_rls_filters",
-        return_value=[],
-    )
 
     assert (
         dataset_macro(1)