diff --git a/authentik/blueprints/management/commands/export_blueprint.py b/authentik/blueprints/management/commands/export_blueprint.py
index f74f1875a..a7a0ccdf8 100644
--- a/authentik/blueprints/management/commands/export_blueprint.py
+++ b/authentik/blueprints/management/commands/export_blueprint.py
@@ -1,24 +1,18 @@
 """Export blueprint of current authentik install"""
-from django.core.management.base import BaseCommand, no_translations
-from django_tenants.management.commands import TenantWrappedCommand
+from django.core.management.base import no_translations
 from structlog.stdlib import get_logger
 
 from authentik.blueprints.v1.exporter import Exporter
+from authentik.tenants.management import TenantCommand
 
 LOGGER = get_logger()
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Export blueprint of current authentik install"""
 
     @no_translations
-    def handle(self, *args, **options):
+    def handle_per_tenant(self, *args, **options):
         """Export blueprint of current authentik install"""
         exporter = Exporter()
         self.stdout.write(exporter.export_to_string())
-
-
-class Command(TenantWrappedCommand):
-    """Export blueprint of current authentik install"""
-
-    COMMAND = TCommand
diff --git a/authentik/crypto/management/commands/import_certificate.py b/authentik/crypto/management/commands/import_certificate.py
index be668e8c2..6f84a1ddc 100644
--- a/authentik/crypto/management/commands/import_certificate.py
+++ b/authentik/crypto/management/commands/import_certificate.py
@@ -1,22 +1,22 @@
 """Import certificate"""
 from sys import exit as sys_exit
 
-from django.core.management.base import BaseCommand, no_translations
-from django_tenants.management.commands import TenantWrappedCommand
+from django.core.management.base import no_translations
 from rest_framework.exceptions import ValidationError
 from structlog.stdlib import get_logger
 
 from authentik.crypto.api import CertificateKeyPairSerializer
 from authentik.crypto.models import CertificateKeyPair
+from authentik.tenants.management import TenantCommand
 
 LOGGER = get_logger()
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Import certificate"""
 
     @no_translations
-    def handle(self, *args, **options):
+    def handle_per_tenant(self, *args, **options):
         """Import certificate"""
         keypair = CertificateKeyPair.objects.filter(name=options["name"]).first()
         dirty = False
@@ -50,9 +50,3 @@ class TCommand(BaseCommand):
         parser.add_argument("--certificate", type=str, required=True)
         parser.add_argument("--private-key", type=str, required=False)
         parser.add_argument("--name", type=str, required=True)
-
-
-class Command(TenantWrappedCommand):
-    """Import certificate"""
-
-    COMMAND = TCommand
diff --git a/authentik/providers/scim/management/commands/scim_sync.py b/authentik/providers/scim/management/commands/scim_sync.py
index 24d10024e..40fa068a6 100644
--- a/authentik/providers/scim/management/commands/scim_sync.py
+++ b/authentik/providers/scim/management/commands/scim_sync.py
@@ -1,30 +1,23 @@
 """SCIM Sync"""
-from django.core.management.base import BaseCommand
-from django_tenants.management.commands import TenantWrappedCommand
 from structlog.stdlib import get_logger
 
 from authentik.providers.scim.models import SCIMProvider
 from authentik.providers.scim.tasks import scim_sync
+from authentik.tenants.management import TenantCommand
 
 LOGGER = get_logger()
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Run sync for an SCIM Provider"""
 
     def add_arguments(self, parser):
         parser.add_argument("providers", nargs="+", type=str)
 
-    def handle(self, **options):
+    def handle_per_tenant(self, **options):
         for provider_name in options["providers"]:
             provider = SCIMProvider.objects.filter(name=provider_name).first()
             if not provider:
                 LOGGER.warning("Provider does not exist", name=provider_name)
                 continue
             scim_sync.delay(provider.pk).get()
-
-
-class Command(TenantWrappedCommand):
-    """Run sync for an SCIM Provider"""
-
-    COMMAND = TCommand
diff --git a/authentik/recovery/management/commands/create_admin_group.py b/authentik/recovery/management/commands/create_admin_group.py
index 26c5ab5be..7605eaceb 100644
--- a/authentik/recovery/management/commands/create_admin_group.py
+++ b/authentik/recovery/management/commands/create_admin_group.py
@@ -1,12 +1,11 @@
 """authentik recovery create_admin_group"""
-from django.core.management.base import BaseCommand
 from django.utils.translation import gettext as _
-from django_tenants.management.commands import TenantWrappedCommand
 
 from authentik.core.models import Group, User
+from authentik.tenants.management import TenantCommand
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Create admin group if the default group gets deleted"""
 
     help = _("Create admin group if the default group gets deleted.")
@@ -14,7 +13,7 @@ class TCommand(BaseCommand):
     def add_arguments(self, parser):
         parser.add_argument("user", action="store", help="User to add to the admin group.")
 
-    def handle(self, *args, **options):
+    def handle_per_tenant(self, *args, **options):
         """Create admin group if the default group gets deleted"""
         username = options.get("user")
         user = User.objects.filter(username=username).first()
@@ -29,9 +28,3 @@ class TCommand(BaseCommand):
         )
         group.users.add(user)
         self.stdout.write(f"User '{username}' successfully added to the group 'authentik Admins'.")
-
-
-class Command(TenantWrappedCommand):
-    """Create admin group if the default group gets deleted"""
-
-    COMMAND = TCommand
diff --git a/authentik/recovery/management/commands/create_recovery_key.py b/authentik/recovery/management/commands/create_recovery_key.py
index 86dec8040..93663dfde 100644
--- a/authentik/recovery/management/commands/create_recovery_key.py
+++ b/authentik/recovery/management/commands/create_recovery_key.py
@@ -2,17 +2,16 @@
 from datetime import timedelta
 from getpass import getuser
 
-from django.core.management.base import BaseCommand
 from django.urls import reverse
 from django.utils.text import slugify
 from django.utils.timezone import now
 from django.utils.translation import gettext as _
-from django_tenants.management.commands import TenantWrappedCommand
 
 from authentik.core.models import Token, TokenIntents, User
+from authentik.tenants.management import TenantCommand
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Create Token used to recover access"""
 
     help = _("Create a Key which can be used to restore access to authentik.")
@@ -30,7 +29,7 @@ class TCommand(BaseCommand):
         """Get full recovery link"""
         return reverse("authentik_recovery:use-token", kwargs={"key": str(token.key)})
 
-    def handle(self, *args, **options):
+    def handle_per_tenant(self, *args, **options):
         """Create Token used to recover access"""
         duration = int(options.get("duration", 1))
         _now = now()
@@ -51,9 +50,3 @@ class TCommand(BaseCommand):
             f"Store this link safely, as it will allow anyone to access authentik as {user}."
         )
         self.stdout.write(self.get_url(token))
-
-
-class Command(TenantWrappedCommand):
-    """Create Token used to recover access"""
-
-    COMMAND = TCommand
diff --git a/authentik/sources/ldap/management/commands/ldap_check_connection.py b/authentik/sources/ldap/management/commands/ldap_check_connection.py
index 2f99bd2f8..c3caed86b 100644
--- a/authentik/sources/ldap/management/commands/ldap_check_connection.py
+++ b/authentik/sources/ldap/management/commands/ldap_check_connection.py
@@ -1,31 +1,24 @@
 """LDAP Connection check"""
 from json import dumps
 
-from django.core.management.base import BaseCommand
-from django_tenants.management.commands import TenantWrappedCommand
 from structlog.stdlib import get_logger
 
 from authentik.sources.ldap.models import LDAPSource
+from authentik.tenants.management import TenantCommand
 
 LOGGER = get_logger()
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Check connectivity to LDAP servers for a source"""
 
     def add_arguments(self, parser):
         parser.add_argument("source_slugs", nargs="?", type=str)
 
-    def handle(self, **options):
+    def handle_per_tenant(self, **options):
         sources = LDAPSource.objects.filter(enabled=True)
         if options["source_slugs"]:
             sources = LDAPSource.objects.filter(slug__in=options["source_slugs"])
         for source in sources.order_by("slug"):
             status = source.check_connection()
             self.stdout.write(dumps(status, indent=4))
-
-
-class Command(TenantWrappedCommand):
-    """Check connectivity to LDAP servers for a source"""
-
-    COMMAND = TCommand
diff --git a/authentik/sources/ldap/management/commands/ldap_sync.py b/authentik/sources/ldap/management/commands/ldap_sync.py
index 67b1ff3ba..fd0a48d63 100644
--- a/authentik/sources/ldap/management/commands/ldap_sync.py
+++ b/authentik/sources/ldap/management/commands/ldap_sync.py
@@ -1,6 +1,4 @@
 """LDAP Sync"""
-from django.core.management.base import BaseCommand
-from django_tenants.management.commands import TenantWrappedCommand
 from structlog.stdlib import get_logger
 
 from authentik.sources.ldap.models import LDAPSource
@@ -8,17 +6,18 @@ from authentik.sources.ldap.sync.groups import GroupLDAPSynchronizer
 from authentik.sources.ldap.sync.membership import MembershipLDAPSynchronizer
 from authentik.sources.ldap.sync.users import UserLDAPSynchronizer
 from authentik.sources.ldap.tasks import ldap_sync_paginator
+from authentik.tenants.management import TenantCommand
 
 LOGGER = get_logger()
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Run sync for an LDAP Source"""
 
     def add_arguments(self, parser):
         parser.add_argument("source_slugs", nargs="+", type=str)
 
-    def handle(self, **options):
+    def handle_per_tenant(self, **options):
         for source_slug in options["source_slugs"]:
             source = LDAPSource.objects.filter(slug=source_slug).first()
             if not source:
@@ -31,9 +30,3 @@ class TCommand(BaseCommand):
             )
             for task in tasks:
                 task()
-
-
-class Command(TenantWrappedCommand):
-    """Run sync for an LDAP Source"""
-
-    COMMAND = TCommand
diff --git a/authentik/stages/email/management/commands/test_email.py b/authentik/stages/email/management/commands/test_email.py
index c26d0526b..fd65b9e5c 100644
--- a/authentik/stages/email/management/commands/test_email.py
+++ b/authentik/stages/email/management/commands/test_email.py
@@ -1,19 +1,19 @@
 """Send a test-email with global settings"""
 from uuid import uuid4
 
-from django.core.management.base import BaseCommand, no_translations
-from django_tenants.management.commands import TenantWrappedCommand
+from django.core.management.base import no_translations
 
 from authentik.stages.email.models import EmailStage
 from authentik.stages.email.tasks import send_mail
 from authentik.stages.email.utils import TemplateEmailMessage
+from authentik.tenants.management import TenantCommand
 
 
-class TCommand(BaseCommand):
+class Command(TenantCommand):
     """Send a test-email with global settings"""
 
     @no_translations
-    def handle(self, *args, **options):
+    def handle_per_tenant(self, *args, **options):
         """Send a test-email with global settings"""
         delete_stage = False
         if options["stage"]:
@@ -42,9 +42,3 @@ class TCommand(BaseCommand):
     def add_arguments(self, parser):
         parser.add_argument("to", type=str)
         parser.add_argument("-S", "--stage", type=str)
-
-
-class Command(TenantWrappedCommand):
-    """Send a test-email with global settings"""
-
-    COMMAND = TCommand
diff --git a/authentik/tenants/management/__init__.py b/authentik/tenants/management/__init__.py
new file mode 100644
index 000000000..7c5cb468d
--- /dev/null
+++ b/authentik/tenants/management/__init__.py
@@ -0,0 +1,40 @@
+from django.core.management.base import BaseCommand
+from django.db import connection
+from django_tenants.utils import get_public_schema_name
+
+from authentik.tenants.models import Tenant
+
+
+class TenantCommand(BaseCommand):
+    """Generic command class useful for running any existing command
+    on a particular tenant."""
+
+    def create_parser(self, prog_name, subcommand, **kwargs):
+        parser = super().create_parser(prog_name, subcommand, **kwargs)
+        self.add_base_argument(
+            parser,
+            "-s",
+            "--schema",
+            default=get_public_schema_name(),
+            help="Tenant schema name.",
+            dest="schema_name",
+        )
+        return parser
+
+    def handle(self, *args, **options):
+        verbosity = int(options.get("verbosity"))
+        schema_name = options["schema_name"] or self.schema_name
+        connection.set_schema_to_public()
+        if verbosity >= 1:
+            self.stderr.write(
+                self.style.NOTICE("Switching to schema '")
+                + self.style.SQL_TABLE(schema_name)
+                + self.style.NOTICE("'")
+            )
+        connection.set_tenant(Tenant.objects.get(schema_name=schema_name))
+        self.handle_per_tenant(*args, **options)
+
+    def handle_per_tenant(self, *args, **options):
+        raise NotImplementedError(
+            "subclasses of TenantCommand must provide a handle_per_tenant() method"
+        )
diff --git a/website/developer-docs/blueprints/export.md b/website/developer-docs/blueprints/export.md
index 94daed668..fa0203197 100644
--- a/website/developer-docs/blueprints/export.md
+++ b/website/developer-docs/blueprints/export.md
@@ -8,7 +8,7 @@ title: Export
 Requires authentik 2022.8.2
 :::
 
-To migrate existing configurations to blueprints, run `ak export_blueprint --schema public` within any authentik Worker container. This will output a blueprint for most currently created objects. Some objects will not be exported as they might have dependencies on other things.
+To migrate existing configurations to blueprints, run `ak export_blueprint` within any authentik Worker container. This will output a blueprint for most currently created objects. Some objects will not be exported as they might have dependencies on other things.
 
 Exported blueprints don't use any of the YAML Tags, they just contain a list of entries as they are in the database.
 
diff --git a/website/docs/core/certificates.md b/website/docs/core/certificates.md
index a1550e8a3..e41763626 100644
--- a/website/docs/core/certificates.md
+++ b/website/docs/core/certificates.md
@@ -62,9 +62,9 @@ Files are checked every 5 minutes, and will trigger an Outpost refresh if the fi
 Starting with authentik 2022.9, you can also import certificates with any folder structure directly. To do this, run the following command within the worker container:
 
 ```shell
-ak import_certificate --schema public --certificate /certs/mycert.pem --private-key /certs/something.pem --name test
+ak import_certificate --certificate /certs/mycert.pem --private-key /certs/something.pem --name test
 # --private-key can be omitted to only import a certificate, i.e. to trust other connections
-# ak import_certificate --schema public --certificate /certs/othercert.pem --name test2
+# ak import_certificate --certificate /certs/othercert.pem --name test2
 ```
 
 This will import the certificate into authentik under the given name. This command is idempotent, meaning you can run it via a cron-job and authentik will only update the certificate when it changes.
diff --git a/website/docs/troubleshooting/emails.md b/website/docs/troubleshooting/emails.md
index ce5aaa475..b3909a49b 100644
--- a/website/docs/troubleshooting/emails.md
+++ b/website/docs/troubleshooting/emails.md
@@ -9,7 +9,7 @@ Some hosting providers block outgoing SMTP ports, in which case you'll have to h
 To test if an email stage, or the global email settings are configured correctly, you can run the following command:
 
 ```
-ak test_email --schema public <to address> [-S <stage name>]
+ak test_email <to address> [-S <stage name>]
 ```
 
 If you omit the `-S` parameter, the email will be sent using the global settings. Otherwise, the settings of the specified stage will be used.
@@ -17,11 +17,11 @@ If you omit the `-S` parameter, the email will be sent using the global settings
 To run this command with docker-compose, use
 
 ```
-docker-compose exec worker ak test_email --schema public [...]
+docker-compose exec worker ak test_email [...]
 ```
 
 To run this command with Kubernetes, use
 
 ```
-kubectl exec -it deployment/authentik-worker -c authentik -- ak test_email --schema public [...]
+kubectl exec -it deployment/authentik-worker -c authentik -- ak test_email [...]
 ```
diff --git a/website/docs/troubleshooting/ldap_source.md b/website/docs/troubleshooting/ldap_source.md
index b00d5adae..ee3f843ef 100644
--- a/website/docs/troubleshooting/ldap_source.md
+++ b/website/docs/troubleshooting/ldap_source.md
@@ -5,23 +5,23 @@ title: Troubleshooting LDAP Synchronization
 To troubleshoot LDAP sources, you can run the command below to run a synchronization in the foreground and see any errors or warnings that might happen directly
 
 ```
-docker-compose run --rm worker ldap_sync --schema public *slug of the source*
+docker-compose run --rm worker ldap_sync *slug of the source*
 ```
 
 or, for Kubernetes, run
 
 ```
-kubectl exec -it deployment/authentik-worker -c authentik -- ak ldap_sync --schema public *slug of the source*
+kubectl exec -it deployment/authentik-worker -c authentik -- ak ldap_sync *slug of the source*
 ```
 
 Starting with authentik 2023.10, you can also run command below to explicitly check the connectivity to the configured LDAP Servers:
 
 ```
-docker-compose run --rm worker ldap_check_connection --schema public *slug of the source*
+docker-compose run --rm worker ldap_check_connection *slug of the source*
 ```
 
 or, for Kubernetes, run
 
 ```
-kubectl exec -it deployment/authentik-worker -c authentik -- ak ldap_check_connection --schema public *slug of the source*
+kubectl exec -it deployment/authentik-worker -c authentik -- ak ldap_check_connection *slug of the source*
 ```
diff --git a/website/docs/troubleshooting/login.md b/website/docs/troubleshooting/login.md
index ffb82137f..58930706d 100644
--- a/website/docs/troubleshooting/login.md
+++ b/website/docs/troubleshooting/login.md
@@ -11,19 +11,19 @@ This recovery key will give whoever has the link direct access to your instances
 To create the key, run the following command:
 
 ```
-docker-compose run --rm server create_recovery_key --schema public 10 akadmin
+docker-compose run --rm server create_recovery_key 10 akadmin
 ```
 
 For Kubernetes, run
 
 ```
-kubectl exec -it deployment/authentik-worker -c authentik -- ak create_recovery_key --schema public 10 akadmin
+kubectl exec -it deployment/authentik-worker -c authentik -- ak create_recovery_key 10 akadmin
 ```
 
 or, for CLI, run
 
 ```
-ak create_recovery_key 10 --schema public akadmin
+ak create_recovery_key 10 akadmin
 ```
 
 This will output a link, that can be used to instantly gain access to authentik as the user specified above. The link is valid for amount of years specified above, in this case, 10 years.
diff --git a/website/docs/troubleshooting/missing_admin_group.md b/website/docs/troubleshooting/missing_admin_group.md
index 2019b463a..a1e612dfd 100644
--- a/website/docs/troubleshooting/missing_admin_group.md
+++ b/website/docs/troubleshooting/missing_admin_group.md
@@ -7,11 +7,11 @@ If all of the Admin groups have been deleted, or misconfigured during sync, you
 Run the following command, where _username_ is the user you want to add to the newly created group:
 
 ```
-docker-compose run --rm server create_admin_group --schema public username
+docker-compose run --rm server create_admin_group username
 ```
 
 or, for Kubernetes, run
 
 ```
-kubectl exec -it deployment/authentik-worker -c authentik -- ak create_admin_group --schema public username
+kubectl exec -it deployment/authentik-worker -c authentik -- ak create_admin_group username
 ```