authentik.blueprints.v1.meta.apply_blueprint

Apply Blueprint meta model

 1"""Apply Blueprint meta model"""
 2
 3from typing import TYPE_CHECKING
 4
 5from rest_framework.exceptions import ValidationError
 6from rest_framework.fields import BooleanField
 7from structlog.stdlib import get_logger
 8
 9from authentik.blueprints.v1.meta.registry import BaseMetaModel, MetaResult, registry
10from authentik.core.api.utils import JSONDictField, PassiveSerializer
11
12if TYPE_CHECKING:
13    from authentik.blueprints.models import BlueprintInstance
14
15LOGGER = get_logger()
16
17
18class ApplyBlueprintMetaSerializer(PassiveSerializer):
19    """Serializer for meta apply blueprint model"""
20
21    identifiers = JSONDictField()
22    required = BooleanField(default=True)
23
24    # We cannot override `instance` as that will confuse rest_framework
25    # and make it attempt to update the instance
26    blueprint_instance: BlueprintInstance
27
28    def validate(self, attrs):
29        from authentik.blueprints.models import BlueprintInstance
30
31        identifiers = attrs["identifiers"]
32        required = attrs["required"]
33        instance = BlueprintInstance.objects.filter(**identifiers).first()
34        if not instance and required:
35            raise ValidationError({"identifiers": "Required blueprint does not exist"})
36        self.blueprint_instance = instance
37        return super().validate(attrs)
38
39    def create(self, validated_data: dict) -> MetaResult:
40        from authentik.blueprints.v1.importer import Importer
41
42        if not self.blueprint_instance:
43            LOGGER.info("Blueprint does not exist, but not required")
44            return MetaResult()
45        LOGGER.debug("Applying blueprint from meta model", blueprint=self.blueprint_instance)
46
47        # Apply blueprint directly using Importer to avoid task context requirements
48        # and prevent deadlocks when called from within another blueprint task
49        blueprint_content = self.blueprint_instance.retrieve()
50        importer = Importer.from_string(blueprint_content, self.blueprint_instance.context)
51        valid, logs = importer.validate()
52        [log.log() for log in logs]
53        if valid:
54            importer.apply()
55        return MetaResult()
56
57
58@registry.register("metaapplyblueprint")
59class MetaApplyBlueprint(BaseMetaModel):
60    """Meta model to apply another blueprint"""
61
62    @staticmethod
63    def serializer() -> ApplyBlueprintMetaSerializer:
64        return ApplyBlueprintMetaSerializer
65
66    class Meta:
67        abstract = True
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
class ApplyBlueprintMetaSerializer(authentik.core.api.utils.PassiveSerializer): 
19class ApplyBlueprintMetaSerializer(PassiveSerializer):
20    """Serializer for meta apply blueprint model"""
21
22    identifiers = JSONDictField()
23    required = BooleanField(default=True)
24
25    # We cannot override `instance` as that will confuse rest_framework
26    # and make it attempt to update the instance
27    blueprint_instance: BlueprintInstance
28
29    def validate(self, attrs):
30        from authentik.blueprints.models import BlueprintInstance
31
32        identifiers = attrs["identifiers"]
33        required = attrs["required"]
34        instance = BlueprintInstance.objects.filter(**identifiers).first()
35        if not instance and required:
36            raise ValidationError({"identifiers": "Required blueprint does not exist"})
37        self.blueprint_instance = instance
38        return super().validate(attrs)
39
40    def create(self, validated_data: dict) -> MetaResult:
41        from authentik.blueprints.v1.importer import Importer
42
43        if not self.blueprint_instance:
44            LOGGER.info("Blueprint does not exist, but not required")
45            return MetaResult()
46        LOGGER.debug("Applying blueprint from meta model", blueprint=self.blueprint_instance)
47
48        # Apply blueprint directly using Importer to avoid task context requirements
49        # and prevent deadlocks when called from within another blueprint task
50        blueprint_content = self.blueprint_instance.retrieve()
51        importer = Importer.from_string(blueprint_content, self.blueprint_instance.context)
52        valid, logs = importer.validate()
53        [log.log() for log in logs]
54        if valid:
55            importer.apply()
56        return MetaResult()

Serializer for meta apply blueprint model

identifiers
required
def validate(self, attrs): 
29    def validate(self, attrs):
30        from authentik.blueprints.models import BlueprintInstance
31
32        identifiers = attrs["identifiers"]
33        required = attrs["required"]
34        instance = BlueprintInstance.objects.filter(**identifiers).first()
35        if not instance and required:
36            raise ValidationError({"identifiers": "Required blueprint does not exist"})
37        self.blueprint_instance = instance
38        return super().validate(attrs)
def create( self, validated_data: dict) -> authentik.blueprints.v1.meta.registry.MetaResult: 
40    def create(self, validated_data: dict) -> MetaResult:
41        from authentik.blueprints.v1.importer import Importer
42
43        if not self.blueprint_instance:
44            LOGGER.info("Blueprint does not exist, but not required")
45            return MetaResult()
46        LOGGER.debug("Applying blueprint from meta model", blueprint=self.blueprint_instance)
47
48        # Apply blueprint directly using Importer to avoid task context requirements
49        # and prevent deadlocks when called from within another blueprint task
50        blueprint_content = self.blueprint_instance.retrieve()
51        importer = Importer.from_string(blueprint_content, self.blueprint_instance.context)
52        valid, logs = importer.validate()
53        [log.log() for log in logs]
54        if valid:
55            importer.apply()
56        return MetaResult()
Inherited Members
authentik.core.api.utils.PassiveSerializer
update
@registry.register('metaapplyblueprint')
class MetaApplyBlueprint(authentik.blueprints.v1.meta.registry.BaseMetaModel): 
59@registry.register("metaapplyblueprint")
60class MetaApplyBlueprint(BaseMetaModel):
61    """Meta model to apply another blueprint"""
62
63    @staticmethod
64    def serializer() -> ApplyBlueprintMetaSerializer:
65        return ApplyBlueprintMetaSerializer
66
67    class Meta:
68        abstract = True

Meta model to apply another blueprint

@staticmethod
def serializer() -> ApplyBlueprintMetaSerializer: 
63    @staticmethod
64    def serializer() -> ApplyBlueprintMetaSerializer:
65        return ApplyBlueprintMetaSerializer

Serializer similar to SerializerModel, but as a static method since this is an abstract model

class MetaApplyBlueprint.Meta: 
67    class Meta:
68        abstract = True
abstract = False