diff --git a/authentik/core/expression/evaluator.py b/authentik/core/expression/evaluator.py
index 18f8ca5e40..23ebeffb77 100644
--- a/authentik/core/expression/evaluator.py
+++ b/authentik/core/expression/evaluator.py
@@ -1,5 +1,6 @@
 """Property Mapping Evaluator"""
 
+from types import CodeType
 from typing import Any
 
 from django.db.models import Model
@@ -24,6 +25,8 @@ class PropertyMappingEvaluator(BaseEvaluator):
     """Custom Evaluator that adds some different context variables."""
 
     dry_run: bool
+    model: Model
+    _compiled: CodeType | None = None
 
     def __init__(
         self,
@@ -33,23 +36,32 @@ class PropertyMappingEvaluator(BaseEvaluator):
         dry_run: bool | None = False,
         **kwargs,
     ):
+        self.model = model
         if hasattr(model, "name"):
             _filename = model.name
         else:
             _filename = str(model)
         super().__init__(filename=_filename)
+        self.dry_run = dry_run
+        self.set_context(user, request, **kwargs)
+
+    def set_context(
+        self,
+        user: User | None = None,
+        request: HttpRequest | None = None,
+        **kwargs,
+    ):
         req = PolicyRequest(user=User())
-        req.obj = model
+        req.obj = self.model
         if user:
             req.user = user
             self._context["user"] = user
         if request:
             req.http_request = request
-        self._context["request"] = req
         req.context.update(**kwargs)
+        self._context["request"] = req
         self._context.update(**kwargs)
         self._globals["SkipObject"] = SkipObjectException
-        self.dry_run = dry_run
 
     def handle_error(self, exc: Exception, expression_source: str):
         """Exception Handler"""
@@ -71,3 +83,9 @@ class PropertyMappingEvaluator(BaseEvaluator):
     def evaluate(self, *args, **kwargs) -> Any:
         with PROPERTY_MAPPING_TIME.labels(mapping_name=self._filename).time():
             return super().evaluate(*args, **kwargs)
+
+    def compile(self, expression: str | None = None) -> Any:
+        if not self._compiled:
+            compiled = super().compile(expression or self.model.expression)
+            self._compiled = compiled
+        return self._compiled
diff --git a/authentik/core/expression/exceptions.py b/authentik/core/expression/exceptions.py
index 210704f0b4..05571e59f4 100644
--- a/authentik/core/expression/exceptions.py
+++ b/authentik/core/expression/exceptions.py
@@ -6,6 +6,11 @@ from authentik.lib.sentry import SentryIgnoredException
 class PropertyMappingExpressionException(SentryIgnoredException):
     """Error when a PropertyMapping Exception expression could not be parsed or evaluated."""
 
+    def __init__(self, exc: Exception, mapping) -> None:
+        super().__init__()
+        self.exc = exc
+        self.mapping = mapping
+
 
 class SkipObjectException(PropertyMappingExpressionException):
     """Exception which can be raised in a property mapping to skip syncing an object.
diff --git a/authentik/core/models.py b/authentik/core/models.py
index a1c17574d8..d12de39905 100644
--- a/authentik/core/models.py
+++ b/authentik/core/models.py
@@ -768,7 +768,7 @@ class PropertyMapping(SerializerModel, ManagedModel):
         try:
             return evaluator.evaluate(self.expression)
         except Exception as exc:
-            raise PropertyMappingExpressionException(exc) from exc
+            raise PropertyMappingExpressionException(self, exc) from exc
 
     def __str__(self):
         return f"Property Mapping {self.name}"
diff --git a/authentik/core/tests/test_models.py b/authentik/core/tests/test_models.py
index 4fd246bbb8..989301a020 100644
--- a/authentik/core/tests/test_models.py
+++ b/authentik/core/tests/test_models.py
@@ -1,10 +1,11 @@
 """authentik core models tests"""
 
 from collections.abc import Callable
-from time import sleep
+from datetime import timedelta
 
 from django.test import RequestFactory, TestCase
 from django.utils.timezone import now
+from freezegun import freeze_time
 from guardian.shortcuts import get_anonymous_user
 
 from authentik.core.models import Provider, Source, Token
@@ -17,15 +18,17 @@ class TestModels(TestCase):
 
     def test_token_expire(self):
         """Test token expiring"""
-        token = Token.objects.create(expires=now(), user=get_anonymous_user())
-        sleep(0.5)
-        self.assertTrue(token.is_expired)
+        with freeze_time() as freeze:
+            token = Token.objects.create(expires=now(), user=get_anonymous_user())
+            freeze.tick(timedelta(seconds=1))
+            self.assertTrue(token.is_expired)
 
     def test_token_expire_no_expire(self):
         """Test token expiring with "expiring" set"""
-        token = Token.objects.create(expires=now(), user=get_anonymous_user(), expiring=False)
-        sleep(0.5)
-        self.assertFalse(token.is_expired)
+        with freeze_time() as freeze:
+            token = Token.objects.create(expires=now(), user=get_anonymous_user(), expiring=False)
+            freeze.tick(timedelta(seconds=1))
+            self.assertFalse(token.is_expired)
 
 
 def source_tester_factory(test_model: type[Stage]) -> Callable:
diff --git a/authentik/enterprise/providers/google_workspace/clients/groups.py b/authentik/enterprise/providers/google_workspace/clients/groups.py
index cde6eaeec4..aeb0ac63e8 100644
--- a/authentik/enterprise/providers/google_workspace/clients/groups.py
+++ b/authentik/enterprise/providers/google_workspace/clients/groups.py
@@ -1,28 +1,22 @@
-from deepmerge import always_merger
 from django.db import transaction
 from django.utils.text import slugify
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import Group
 from authentik.enterprise.providers.google_workspace.clients.base import GoogleWorkspaceSyncClient
 from authentik.enterprise.providers.google_workspace.models import (
+    GoogleWorkspaceProvider,
     GoogleWorkspaceProviderGroup,
     GoogleWorkspaceProviderMapping,
     GoogleWorkspaceProviderUser,
 )
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.base import Direction
 from authentik.lib.sync.outgoing.exceptions import (
     NotFoundSyncException,
     ObjectExistsSyncException,
-    StopSync,
     TransientSyncException,
 )
 from authentik.lib.sync.outgoing.models import OutgoingSyncDeleteAction
-from authentik.lib.utils.errors import exception_to_string
 
 
 class GoogleWorkspaceGroupClient(
@@ -34,41 +28,21 @@ class GoogleWorkspaceGroupClient(
     connection_type_query = "group"
     can_discover = True
 
+    def __init__(self, provider: GoogleWorkspaceProvider) -> None:
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings_group.all().order_by("name").select_subclasses(),
+            GoogleWorkspaceProviderMapping,
+            ["group", "provider", "creating"],
+        )
+
     def to_schema(self, obj: Group, creating: bool) -> dict:
         """Convert authentik group"""
-        raw_google_group = {}
-        for mapping in (
-            self.provider.property_mappings_group.all().order_by("name").select_subclasses()
-        ):
-            if not isinstance(mapping, GoogleWorkspaceProviderMapping):
-                continue
-            try:
-                value = mapping.evaluate(
-                    user=None,
-                    request=None,
-                    group=obj,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_google_group, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_google_group:
-            raise StopSync(ValueError("No group mappings configured"), obj)
-        raw_google_group.setdefault(
-            "email", f"{slugify(obj.name)}@{self.provider.default_group_email_domain}"
+        return super().to_schema(
+            obj,
+            creating,
+            email=f"{slugify(obj.name)}@{self.provider.default_group_email_domain}",
         )
-        return raw_google_group
 
     def delete(self, obj: Group):
         """Delete group"""
diff --git a/authentik/enterprise/providers/google_workspace/clients/users.py b/authentik/enterprise/providers/google_workspace/clients/users.py
index 03db3eded7..52d60046bd 100644
--- a/authentik/enterprise/providers/google_workspace/clients/users.py
+++ b/authentik/enterprise/providers/google_workspace/clients/users.py
@@ -1,24 +1,18 @@
-from deepmerge import always_merger
 from django.db import transaction
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import User
 from authentik.enterprise.providers.google_workspace.clients.base import GoogleWorkspaceSyncClient
 from authentik.enterprise.providers.google_workspace.models import (
+    GoogleWorkspaceProvider,
     GoogleWorkspaceProviderMapping,
     GoogleWorkspaceProviderUser,
 )
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.exceptions import (
     ObjectExistsSyncException,
-    StopSync,
     TransientSyncException,
 )
 from authentik.lib.sync.outgoing.models import OutgoingSyncDeleteAction
-from authentik.lib.utils.errors import exception_to_string
 from authentik.policies.utils import delete_none_values
 
 
@@ -29,35 +23,19 @@ class GoogleWorkspaceUserClient(GoogleWorkspaceSyncClient[User, GoogleWorkspaceP
     connection_type_query = "user"
     can_discover = True
 
+    def __init__(self, provider: GoogleWorkspaceProvider) -> None:
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings.all().order_by("name").select_subclasses(),
+            GoogleWorkspaceProviderMapping,
+            ["provider", "creating"],
+        )
+
     def to_schema(self, obj: User, creating: bool) -> dict:
         """Convert authentik user"""
-        raw_google_user = {}
-        for mapping in self.provider.property_mappings.all().order_by("name").select_subclasses():
-            if not isinstance(mapping, GoogleWorkspaceProviderMapping):
-                continue
-            try:
-                value = mapping.evaluate(
-                    user=obj,
-                    request=None,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_google_user, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_google_user:
-            raise StopSync(ValueError("No user mappings configured"), obj)
-        raw_google_user.setdefault("primaryEmail", str(obj.email))
+        raw_google_user = super().to_schema(obj, creating)
+        if "primaryEmail" not in raw_google_user:
+            raw_google_user["primaryEmail"] = str(obj.email)
         return delete_none_values(raw_google_user)
 
     def delete(self, obj: User):
diff --git a/authentik/enterprise/providers/microsoft_entra/clients/groups.py b/authentik/enterprise/providers/microsoft_entra/clients/groups.py
index eba5400998..0121c132be 100644
--- a/authentik/enterprise/providers/microsoft_entra/clients/groups.py
+++ b/authentik/enterprise/providers/microsoft_entra/clients/groups.py
@@ -1,21 +1,17 @@
-from deepmerge import always_merger
 from django.db import transaction
 from msgraph.generated.groups.groups_request_builder import GroupsRequestBuilder
 from msgraph.generated.models.group import Group as MSGroup
 from msgraph.generated.models.reference_create import ReferenceCreate
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import Group
 from authentik.enterprise.providers.microsoft_entra.clients.base import MicrosoftEntraSyncClient
 from authentik.enterprise.providers.microsoft_entra.models import (
+    MicrosoftEntraProvider,
     MicrosoftEntraProviderGroup,
     MicrosoftEntraProviderMapping,
     MicrosoftEntraProviderUser,
 )
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.base import Direction
 from authentik.lib.sync.outgoing.exceptions import (
     NotFoundSyncException,
@@ -24,7 +20,6 @@ from authentik.lib.sync.outgoing.exceptions import (
     TransientSyncException,
 )
 from authentik.lib.sync.outgoing.models import OutgoingSyncDeleteAction
-from authentik.lib.utils.errors import exception_to_string
 
 
 class MicrosoftEntraGroupClient(
@@ -36,37 +31,17 @@ class MicrosoftEntraGroupClient(
     connection_type_query = "group"
     can_discover = True
 
+    def __init__(self, provider: MicrosoftEntraProvider) -> None:
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings_group.all().order_by("name").select_subclasses(),
+            MicrosoftEntraProviderMapping,
+            ["group", "provider", "creating"],
+        )
+
     def to_schema(self, obj: Group, creating: bool) -> MSGroup:
         """Convert authentik group"""
-        raw_microsoft_group = {}
-        for mapping in (
-            self.provider.property_mappings_group.all().order_by("name").select_subclasses()
-        ):
-            if not isinstance(mapping, MicrosoftEntraProviderMapping):
-                continue
-            try:
-                value = mapping.evaluate(
-                    user=None,
-                    request=None,
-                    group=obj,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_microsoft_group, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_microsoft_group:
-            raise StopSync(ValueError("No group mappings configured"), obj)
+        raw_microsoft_group = super().to_schema(obj, creating)
         try:
             return MSGroup(**raw_microsoft_group)
         except TypeError as exc:
diff --git a/authentik/enterprise/providers/microsoft_entra/clients/users.py b/authentik/enterprise/providers/microsoft_entra/clients/users.py
index 99c682175e..a9539ba465 100644
--- a/authentik/enterprise/providers/microsoft_entra/clients/users.py
+++ b/authentik/enterprise/providers/microsoft_entra/clients/users.py
@@ -1,26 +1,21 @@
-from deepmerge import always_merger
 from django.db import transaction
 from msgraph.generated.models.user import User as MSUser
 from msgraph.generated.users.users_request_builder import UsersRequestBuilder
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import User
 from authentik.enterprise.providers.microsoft_entra.clients.base import MicrosoftEntraSyncClient
 from authentik.enterprise.providers.microsoft_entra.models import (
+    MicrosoftEntraProvider,
     MicrosoftEntraProviderMapping,
     MicrosoftEntraProviderUser,
 )
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.exceptions import (
     ObjectExistsSyncException,
     StopSync,
     TransientSyncException,
 )
 from authentik.lib.sync.outgoing.models import OutgoingSyncDeleteAction
-from authentik.lib.utils.errors import exception_to_string
 from authentik.policies.utils import delete_none_values
 
 
@@ -31,34 +26,17 @@ class MicrosoftEntraUserClient(MicrosoftEntraSyncClient[User, MicrosoftEntraProv
     connection_type_query = "user"
     can_discover = True
 
+    def __init__(self, provider: MicrosoftEntraProvider) -> None:
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings.all().order_by("name").select_subclasses(),
+            MicrosoftEntraProviderMapping,
+            ["provider", "creating"],
+        )
+
     def to_schema(self, obj: User, creating: bool) -> MSUser:
         """Convert authentik user"""
-        raw_microsoft_user = {}
-        for mapping in self.provider.property_mappings.all().order_by("name").select_subclasses():
-            if not isinstance(mapping, MicrosoftEntraProviderMapping):
-                continue
-            try:
-                value = mapping.evaluate(
-                    user=obj,
-                    request=None,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_microsoft_user, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_microsoft_user:
-            raise StopSync(ValueError("No user mappings configured"), obj)
+        raw_microsoft_user = super().to_schema(obj, creating)
         try:
             return MSUser(**delete_none_values(raw_microsoft_user))
         except TypeError as exc:
diff --git a/authentik/lib/expression/evaluator.py b/authentik/lib/expression/evaluator.py
index e61a429fcf..f17b346d90 100644
--- a/authentik/lib/expression/evaluator.py
+++ b/authentik/lib/expression/evaluator.py
@@ -5,6 +5,7 @@ import socket
 from collections.abc import Iterable
 from ipaddress import ip_address, ip_network
 from textwrap import indent
+from types import CodeType
 from typing import Any
 
 from cachetools import TLRUCache, cached
@@ -184,7 +185,7 @@ class BaseEvaluator:
         full_expression += f"\nresult = handler({handler_signature})"
         return full_expression
 
-    def compile(self, expression: str) -> Any:
+    def compile(self, expression: str) -> CodeType:
         """Parse expression. Raises SyntaxError or ValueError if the syntax is incorrect."""
         param_keys = self._context.keys()
         return compile(self.wrap_expression(expression, param_keys), self._filename, "exec")
diff --git a/authentik/lib/sync/mapper.py b/authentik/lib/sync/mapper.py
new file mode 100644
index 0000000000..18211a1104
--- /dev/null
+++ b/authentik/lib/sync/mapper.py
@@ -0,0 +1,67 @@
+from collections.abc import Generator
+
+from django.db.models import QuerySet
+from django.http import HttpRequest
+
+from authentik.core.expression.evaluator import PropertyMappingEvaluator
+from authentik.core.expression.exceptions import PropertyMappingExpressionException
+from authentik.core.models import PropertyMapping, User
+
+
+class PropertyMappingManager:
+    """Pre-compile and cache property mappings when an identical
+    set is used multiple times"""
+
+    query_set: QuerySet[PropertyMapping]
+    mapping_subclass: type[PropertyMapping]
+
+    _evaluators: list[PropertyMappingEvaluator]
+
+    def __init__(
+        self,
+        qs: QuerySet[PropertyMapping],
+        # Expected subclass of PropertyMappings, any objects in the queryset
+        # that are not an instance of this class will be discarded
+        mapping_subclass: type[PropertyMapping],
+        # As they keys of parameters are part of the compilation,
+        # we need a list of all parameter names that will be used during evaluation
+        context_keys: list[str],
+    ) -> None:
+        self.query_set = qs
+        self.mapping_subclass = mapping_subclass
+        self.context_keys = context_keys
+        self.compile()
+
+    def compile(self):
+        self._evaluators = []
+        for mapping in self.query_set:
+            if not isinstance(mapping, self.mapping_subclass):
+                continue
+            evaluator = PropertyMappingEvaluator(
+                mapping, **{key: None for key in self.context_keys}
+            )
+            # Compile and cache expression
+            evaluator.compile()
+            self._evaluators.append(evaluator)
+
+    def iter_eval(
+        self,
+        user: User | None,
+        request: HttpRequest | None,
+        return_mapping: bool = False,
+        **kwargs,
+    ) -> Generator[tuple[dict, PropertyMapping], None]:
+        """Iterate over all mappings that were pre-compiled and
+        execute all of them with the given context"""
+        for mapping in self._evaluators:
+            mapping.set_context(user, request, **kwargs)
+            try:
+                value = mapping.evaluate(mapping.model.expression)
+            except Exception as exc:
+                raise PropertyMappingExpressionException(exc, mapping.model) from exc
+            if value is None:
+                continue
+            if return_mapping:
+                yield value, mapping.model
+            else:
+                yield value
diff --git a/authentik/lib/sync/outgoing/base.py b/authentik/lib/sync/outgoing/base.py
index cb04bb3293..45b60cd2ce 100644
--- a/authentik/lib/sync/outgoing/base.py
+++ b/authentik/lib/sync/outgoing/base.py
@@ -3,10 +3,18 @@
 from enum import StrEnum
 from typing import TYPE_CHECKING
 
+from deepmerge import always_merger
 from django.db import DatabaseError
 from structlog.stdlib import get_logger
 
-from authentik.lib.sync.outgoing.exceptions import NotFoundSyncException
+from authentik.core.expression.exceptions import (
+    PropertyMappingExpressionException,
+    SkipObjectException,
+)
+from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
+from authentik.lib.sync.outgoing.exceptions import NotFoundSyncException, StopSync
+from authentik.lib.utils.errors import exception_to_string
 
 if TYPE_CHECKING:
     from django.db.models import Model
@@ -28,6 +36,7 @@ class BaseOutgoingSyncClient[
     provider: TProvider
     connection_type: type[TConnection]
     connection_type_query: str
+    mapper: PropertyMappingManager
 
     can_discover = False
 
@@ -70,9 +79,35 @@ class BaseOutgoingSyncClient[
         """Delete object from destination"""
         raise NotImplementedError()
 
-    def to_schema(self, obj: TModel, creating: bool) -> TSchema:
+    def to_schema(self, obj: TModel, creating: bool, **defaults) -> TSchema:
         """Convert object to destination schema"""
-        raise NotImplementedError()
+        raw_final_object = {}
+        try:
+            eval_kwargs = {
+                "request": None,
+                "provider": self.provider,
+                "creating": creating,
+                obj._meta.model_name: obj,
+            }
+            eval_kwargs.setdefault("user", None)
+            for value in self.mapper.iter_eval(**eval_kwargs):
+                try:
+                    always_merger.merge(raw_final_object, value)
+                except SkipObjectException as exc:
+                    raise exc from exc
+        except PropertyMappingExpressionException as exc:
+            # Value error can be raised when assigning invalid data to an attribute
+            Event.new(
+                EventAction.CONFIGURATION_ERROR,
+                message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
+                mapping=exc.mapping,
+            ).save()
+            raise StopSync(exc, obj, exc.mapping) from exc
+        if not raw_final_object:
+            raise StopSync(ValueError("No user mappings configured"), obj)
+        for key, value in defaults.items():
+            raw_final_object.setdefault(key, value)
+        return raw_final_object
 
     def discover(self):
         """Optional method. Can be used to implement a "discovery" where
diff --git a/authentik/providers/scim/clients/groups.py b/authentik/providers/scim/clients/groups.py
index 7b95d11846..dc4fadb74e 100644
--- a/authentik/providers/scim/clients/groups.py
+++ b/authentik/providers/scim/clients/groups.py
@@ -1,31 +1,25 @@
 """Group client"""
 
-from deepmerge import always_merger
 from pydantic import ValidationError
 from pydanticscim.group import GroupMember
 from pydanticscim.responses import PatchOp, PatchOperation
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import Group
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.base import Direction
 from authentik.lib.sync.outgoing.exceptions import (
     NotFoundSyncException,
     ObjectExistsSyncException,
     StopSync,
 )
-from authentik.lib.utils.errors import exception_to_string
 from authentik.policies.utils import delete_none_values
 from authentik.providers.scim.clients.base import SCIMClient
 from authentik.providers.scim.clients.exceptions import (
     SCIMRequestException,
 )
+from authentik.providers.scim.clients.schema import SCIM_GROUP_SCHEMA, PatchRequest
 from authentik.providers.scim.clients.schema import Group as SCIMGroupSchema
-from authentik.providers.scim.clients.schema import PatchRequest
-from authentik.providers.scim.models import SCIMGroup, SCIMMapping, SCIMUser
+from authentik.providers.scim.models import SCIMGroup, SCIMMapping, SCIMProvider, SCIMUser
 
 
 class SCIMGroupClient(SCIMClient[Group, SCIMGroup, SCIMGroupSchema]):
@@ -33,41 +27,23 @@ class SCIMGroupClient(SCIMClient[Group, SCIMGroup, SCIMGroupSchema]):
 
     connection_type = SCIMGroup
     connection_type_query = "group"
+    mapper: PropertyMappingManager
+
+    def __init__(self, provider: SCIMProvider):
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings_group.all().order_by("name").select_subclasses(),
+            SCIMMapping,
+            ["group", "provider", "creating"],
+        )
 
     def to_schema(self, obj: Group, creating: bool) -> SCIMGroupSchema:
         """Convert authentik user into SCIM"""
-        raw_scim_group = {
-            "schemas": ("urn:ietf:params:scim:schemas:core:2.0:Group",),
-        }
-        for mapping in (
-            self.provider.property_mappings_group.all().order_by("name").select_subclasses()
-        ):
-            if not isinstance(mapping, SCIMMapping):
-                continue
-            try:
-                mapping: SCIMMapping
-                value = mapping.evaluate(
-                    user=None,
-                    request=None,
-                    group=obj,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_scim_group, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_scim_group:
-            raise StopSync(ValueError("No group mappings configured"), obj)
+        raw_scim_group = super().to_schema(
+            obj,
+            creating,
+            schemas=(SCIM_GROUP_SCHEMA,),
+        )
         try:
             scim_group = SCIMGroupSchema.model_validate(delete_none_values(raw_scim_group))
         except ValidationError as exc:
diff --git a/authentik/providers/scim/clients/users.py b/authentik/providers/scim/clients/users.py
index 075559a766..350020d34f 100644
--- a/authentik/providers/scim/clients/users.py
+++ b/authentik/providers/scim/clients/users.py
@@ -1,20 +1,15 @@
 """User client"""
 
-from deepmerge import always_merger
 from pydantic import ValidationError
 
-from authentik.core.expression.exceptions import (
-    PropertyMappingExpressionException,
-    SkipObjectException,
-)
 from authentik.core.models import User
-from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
 from authentik.lib.sync.outgoing.exceptions import StopSync
-from authentik.lib.utils.errors import exception_to_string
 from authentik.policies.utils import delete_none_values
 from authentik.providers.scim.clients.base import SCIMClient
+from authentik.providers.scim.clients.schema import SCIM_USER_SCHEMA
 from authentik.providers.scim.clients.schema import User as SCIMUserSchema
-from authentik.providers.scim.models import SCIMMapping, SCIMUser
+from authentik.providers.scim.models import SCIMMapping, SCIMProvider, SCIMUser
 
 
 class SCIMUserClient(SCIMClient[User, SCIMUser, SCIMUserSchema]):
@@ -22,38 +17,23 @@ class SCIMUserClient(SCIMClient[User, SCIMUser, SCIMUserSchema]):
 
     connection_type = SCIMUser
     connection_type_query = "user"
+    mapper: PropertyMappingManager
+
+    def __init__(self, provider: SCIMProvider):
+        super().__init__(provider)
+        self.mapper = PropertyMappingManager(
+            self.provider.property_mappings.all().order_by("name").select_subclasses(),
+            SCIMMapping,
+            ["provider", "creating"],
+        )
 
     def to_schema(self, obj: User, creating: bool) -> SCIMUserSchema:
         """Convert authentik user into SCIM"""
-        raw_scim_user = {
-            "schemas": ("urn:ietf:params:scim:schemas:core:2.0:User",),
-        }
-        for mapping in self.provider.property_mappings.all().order_by("name").select_subclasses():
-            if not isinstance(mapping, SCIMMapping):
-                continue
-            try:
-                mapping: SCIMMapping
-                value = mapping.evaluate(
-                    user=obj,
-                    request=None,
-                    provider=self.provider,
-                    creating=creating,
-                )
-                if value is None:
-                    continue
-                always_merger.merge(raw_scim_user, value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except (PropertyMappingExpressionException, ValueError) as exc:
-                # Value error can be raised when assigning invalid data to an attribute
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
-                    mapping=mapping,
-                ).save()
-                raise StopSync(exc, obj, mapping) from exc
-        if not raw_scim_user:
-            raise StopSync(ValueError("No user mappings configured"), obj)
+        raw_scim_user = super().to_schema(
+            obj,
+            creating,
+            schemas=(SCIM_USER_SCHEMA,),
+        )
         try:
             scim_user = SCIMUserSchema.model_validate(delete_none_values(raw_scim_user))
         except ValidationError as exc:
diff --git a/authentik/sources/ldap/sync/base.py b/authentik/sources/ldap/sync/base.py
index 089690d4bc..8435b80d5c 100644
--- a/authentik/sources/ldap/sync/base.py
+++ b/authentik/sources/ldap/sync/base.py
@@ -5,7 +5,6 @@ from typing import Any
 
 from django.conf import settings
 from django.db.models.base import Model
-from django.db.models.query import QuerySet
 from ldap3 import DEREF_ALWAYS, SUBTREE, Connection
 from structlog.stdlib import BoundLogger, get_logger
 
@@ -16,8 +15,11 @@ from authentik.core.expression.exceptions import (
 from authentik.events.models import Event, EventAction
 from authentik.lib.config import CONFIG, set_path_in_dict
 from authentik.lib.merge import MERGE_LIST_UNIQUE
+from authentik.lib.sync.mapper import PropertyMappingManager
+from authentik.lib.sync.outgoing.exceptions import StopSync
+from authentik.lib.utils.errors import exception_to_string
 from authentik.sources.ldap.auth import LDAP_DISTINGUISHED_NAME
-from authentik.sources.ldap.models import LDAPPropertyMapping, LDAPSource
+from authentik.sources.ldap.models import LDAPSource
 
 LDAP_UNIQUENESS = "ldap_uniq"
 
@@ -38,6 +40,7 @@ class BaseLDAPSynchronizer:
     _logger: BoundLogger
     _connection: Connection
     _messages: list[str]
+    mapper: PropertyMappingManager
 
     def __init__(self, source: LDAPSource):
         self._source = source
@@ -139,52 +142,47 @@ class BaseLDAPSynchronizer:
 
     def build_user_properties(self, user_dn: str, **kwargs) -> dict[str, Any]:
         """Build attributes for User object based on property mappings."""
-        props = self._build_object_properties(user_dn, self._source.property_mappings, **kwargs)
+        props = self._build_object_properties(user_dn, **kwargs)
         props.setdefault("path", self._source.get_user_path())
         return props
 
     def build_group_properties(self, group_dn: str, **kwargs) -> dict[str, Any]:
         """Build attributes for Group object based on property mappings."""
-        return self._build_object_properties(
-            group_dn, self._source.property_mappings_group, **kwargs
-        )
+        return self._build_object_properties(group_dn, **kwargs)
 
-    def _build_object_properties(
-        self, object_dn: str, mappings: QuerySet, **kwargs
-    ) -> dict[str, dict[Any, Any]]:
+    def _build_object_properties(self, object_dn: str, **kwargs) -> dict[str, dict[Any, Any]]:
         properties = {"attributes": {}}
-        for mapping in mappings.all().select_subclasses():
-            if not isinstance(mapping, LDAPPropertyMapping):
-                continue
-            mapping: LDAPPropertyMapping
-            try:
-                value = mapping.evaluate(
-                    user=None, request=None, ldap=kwargs, dn=object_dn, source=self._source
-                )
-                if value is None:
-                    self._logger.warning("property mapping returned None", mapping=mapping)
-                    continue
-                if isinstance(value, (bytes)):
-                    self._logger.warning("property mapping returned bytes", mapping=mapping)
-                    continue
-                object_field = mapping.object_field
-                if object_field.startswith("attributes."):
-                    # Because returning a list might desired, we can't
-                    # rely on flatten here. Instead, just save the result as-is
-                    set_path_in_dict(properties, object_field, value)
-                else:
-                    properties[object_field] = flatten(value)
-            except SkipObjectException as exc:
-                raise exc from exc
-            except PropertyMappingExpressionException as exc:
-                Event.new(
-                    EventAction.CONFIGURATION_ERROR,
-                    message=f"Failed to evaluate property-mapping: '{mapping.name}'",
-                    source=self._source,
-                    mapping=mapping,
-                ).save()
-                self._logger.warning("Mapping failed to evaluate", exc=exc, mapping=mapping)
-                continue
+        try:
+            for value, mapping in self.mapper.iter_eval(
+                user=None,
+                request=None,
+                return_mapping=True,
+                ldap=kwargs,
+                dn=object_dn,
+                source=self._source,
+            ):
+                try:
+                    if isinstance(value, (bytes)):
+                        self._logger.warning("property mapping returned bytes", mapping=mapping)
+                        continue
+                    object_field = mapping.object_field
+                    if object_field.startswith("attributes."):
+                        # Because returning a list might desired, we can't
+                        # rely on flatten here. Instead, just save the result as-is
+                        set_path_in_dict(properties, object_field, value)
+                    else:
+                        properties[object_field] = flatten(value)
+                except SkipObjectException as exc:
+                    raise exc from exc
+        except PropertyMappingExpressionException as exc:
+            # Value error can be raised when assigning invalid data to an attribute
+            Event.new(
+                EventAction.CONFIGURATION_ERROR,
+                message=f"Failed to evaluate property-mapping {exception_to_string(exc)}",
+                mapping=exc.mapping,
+            ).save()
+            self._logger.warning("Mapping failed to evaluate", exc=exc, mapping=exc.mapping)
+            raise StopSync(exc, None, exc.mapping) from exc
         if self._source.object_uniqueness_field in kwargs:
             properties["attributes"][LDAP_UNIQUENESS] = flatten(
                 kwargs.get(self._source.object_uniqueness_field)
diff --git a/authentik/sources/ldap/sync/groups.py b/authentik/sources/ldap/sync/groups.py
index d239c196bb..739e1af911 100644
--- a/authentik/sources/ldap/sync/groups.py
+++ b/authentik/sources/ldap/sync/groups.py
@@ -9,12 +9,22 @@ from ldap3 import ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, SUBTREE
 from authentik.core.expression.exceptions import SkipObjectException
 from authentik.core.models import Group
 from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
+from authentik.sources.ldap.models import LDAPPropertyMapping, LDAPSource
 from authentik.sources.ldap.sync.base import LDAP_UNIQUENESS, BaseLDAPSynchronizer, flatten
 
 
 class GroupLDAPSynchronizer(BaseLDAPSynchronizer):
     """Sync LDAP Users and groups into authentik"""
 
+    def __init__(self, source: LDAPSource):
+        super().__init__(source)
+        self.mapper = PropertyMappingManager(
+            self._source.property_mappings_group.all().order_by("name").select_subclasses(),
+            LDAPPropertyMapping,
+            ["ldap", "dn", "source"],
+        )
+
     @staticmethod
     def name() -> str:
         return "groups"
diff --git a/authentik/sources/ldap/sync/users.py b/authentik/sources/ldap/sync/users.py
index 4ccfe47136..188dad1771 100644
--- a/authentik/sources/ldap/sync/users.py
+++ b/authentik/sources/ldap/sync/users.py
@@ -9,6 +9,8 @@ from ldap3 import ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, SUBTREE
 from authentik.core.expression.exceptions import SkipObjectException
 from authentik.core.models import User
 from authentik.events.models import Event, EventAction
+from authentik.lib.sync.mapper import PropertyMappingManager
+from authentik.sources.ldap.models import LDAPPropertyMapping, LDAPSource
 from authentik.sources.ldap.sync.base import LDAP_UNIQUENESS, BaseLDAPSynchronizer, flatten
 from authentik.sources.ldap.sync.vendor.freeipa import FreeIPA
 from authentik.sources.ldap.sync.vendor.ms_ad import MicrosoftActiveDirectory
@@ -17,6 +19,14 @@ from authentik.sources.ldap.sync.vendor.ms_ad import MicrosoftActiveDirectory
 class UserLDAPSynchronizer(BaseLDAPSynchronizer):
     """Sync LDAP Users into authentik"""
 
+    def __init__(self, source: LDAPSource):
+        super().__init__(source)
+        self.mapper = PropertyMappingManager(
+            self._source.property_mappings.all().order_by("name").select_subclasses(),
+            LDAPPropertyMapping,
+            ["ldap", "dn", "source"],
+        )
+
     @staticmethod
     def name() -> str:
         return "users"
diff --git a/authentik/sources/ldap/tasks.py b/authentik/sources/ldap/tasks.py
index 9184089b97..356ffa9f40 100644
--- a/authentik/sources/ldap/tasks.py
+++ b/authentik/sources/ldap/tasks.py
@@ -12,6 +12,7 @@ from authentik.events.models import SystemTask as DBSystemTask
 from authentik.events.models import TaskStatus
 from authentik.events.system_tasks import SystemTask
 from authentik.lib.config import CONFIG
+from authentik.lib.sync.outgoing.exceptions import StopSync
 from authentik.lib.utils.errors import exception_to_string
 from authentik.lib.utils.reflection import class_to_path, path_to_class
 from authentik.root.celery import CELERY_APP
@@ -138,7 +139,7 @@ def ldap_sync(self: SystemTask, source_pk: str, sync_class: str, page_cache_key:
             *messages,
         )
         cache.delete(page_cache_key)
-    except LDAPException as exc:
+    except (LDAPException, StopSync) as exc:
         # No explicit event is created here as .set_status with an error will do that
         LOGGER.warning(exception_to_string(exc))
         self.set_error(exc)
diff --git a/authentik/sources/ldap/tests/test_sync.py b/authentik/sources/ldap/tests/test_sync.py
index 8ccca9f906..1c171f2323 100644
--- a/authentik/sources/ldap/tests/test_sync.py
+++ b/authentik/sources/ldap/tests/test_sync.py
@@ -11,6 +11,7 @@ from authentik.core.tests.utils import create_test_admin_user
 from authentik.events.models import Event, EventAction, SystemTask
 from authentik.events.system_tasks import TaskStatus
 from authentik.lib.generators import generate_id, generate_key
+from authentik.lib.sync.outgoing.exceptions import StopSync
 from authentik.lib.utils.reflection import class_to_path
 from authentik.sources.ldap.models import LDAPPropertyMapping, LDAPSource
 from authentik.sources.ldap.sync.groups import GroupLDAPSynchronizer
@@ -63,12 +64,13 @@ class LDAPSyncTests(TestCase):
         connection = MagicMock(return_value=mock_ad_connection(LDAP_PASSWORD))
         with patch("authentik.sources.ldap.models.LDAPSource.connection", connection):
             user_sync = UserLDAPSynchronizer(self.source)
-            user_sync.sync_full()
+            with self.assertRaises(StopSync):
+                user_sync.sync_full()
             self.assertFalse(User.objects.filter(username="user0_sn").exists())
             self.assertFalse(User.objects.filter(username="user1_sn").exists())
         events = Event.objects.filter(
             action=EventAction.CONFIGURATION_ERROR,
-            context__message="Failed to evaluate property-mapping: 'name'",
+            context__mapping__pk=mapping.pk.hex,
         )
         self.assertTrue(events.exists())
 
diff --git a/authentik/stages/consent/tests.py b/authentik/stages/consent/tests.py
index f67496482a..872539ad24 100644
--- a/authentik/stages/consent/tests.py
+++ b/authentik/stages/consent/tests.py
@@ -1,8 +1,9 @@
 """consent tests"""
 
-from time import sleep
+from datetime import timedelta
 
 from django.urls import reverse
+from freezegun import freeze_time
 
 from authentik.core.models import Application
 from authentik.core.tasks import clean_expired_models
@@ -136,11 +137,12 @@ class TestConsentStage(FlowTestCase):
         self.assertTrue(
             UserConsent.objects.filter(user=self.user, application=self.application).exists()
         )
-        sleep(1)
-        clean_expired_models.delay().get()
-        self.assertFalse(
-            UserConsent.objects.filter(user=self.user, application=self.application).exists()
-        )
+        with freeze_time() as frozen_time:
+            frozen_time.tick(timedelta(seconds=3))
+            clean_expired_models.delay().get()
+            self.assertFalse(
+                UserConsent.objects.filter(user=self.user, application=self.application).exists()
+            )
 
     def test_permanent_more_perms(self):
         """Test permanent consent from user"""
diff --git a/authentik/stages/prompt/models.py b/authentik/stages/prompt/models.py
index 4e11451243..1a4a6c1123 100644
--- a/authentik/stages/prompt/models.py
+++ b/authentik/stages/prompt/models.py
@@ -208,7 +208,7 @@ class Prompt(SerializerModel):
             try:
                 return evaluator.evaluate(self.placeholder)
             except Exception as exc:  # pylint:disable=broad-except
-                wrapped = PropertyMappingExpressionException(str(exc))
+                wrapped = PropertyMappingExpressionException(str(exc), None)
                 LOGGER.warning(
                     "failed to evaluate prompt placeholder",
                     exc=wrapped,