authentik.flows.api.stages
Flow Stage API Views
1"""Flow Stage API Views""" 2 3from uuid import uuid4 4 5from django.urls.base import reverse 6from drf_spectacular.utils import extend_schema 7from rest_framework import mixins 8from rest_framework.decorators import action 9from rest_framework.fields import SerializerMethodField 10from rest_framework.request import Request 11from rest_framework.response import Response 12from rest_framework.viewsets import GenericViewSet 13from structlog.stdlib import get_logger 14 15from authentik.core.api.object_types import TypesMixin 16from authentik.core.api.used_by import UsedByMixin 17from authentik.core.api.utils import MetaNameSerializer, ModelSerializer 18from authentik.core.types import UserSettingSerializer 19from authentik.flows.api.flows import FlowSetSerializer 20from authentik.flows.models import ConfigurableStage, Stage 21from authentik.lib.utils.reflection import all_subclasses 22 23LOGGER = get_logger() 24 25 26class StageSerializer(ModelSerializer, MetaNameSerializer): 27 """Stage Serializer""" 28 29 component = SerializerMethodField() 30 flow_set = FlowSetSerializer(many=True, required=False, read_only=True) 31 32 def to_representation(self, instance: Stage): 33 if isinstance(instance, Stage) and instance.is_in_memory: 34 instance.stage_uuid = uuid4() 35 return super().to_representation(instance) 36 37 def get_component(self, obj: Stage) -> str: 38 """Get object type so that we know how to edit the object""" 39 if obj.__class__ == Stage: 40 return "" 41 return obj.component 42 43 class Meta: 44 model = Stage 45 fields = [ 46 "pk", 47 "name", 48 "component", 49 "verbose_name", 50 "verbose_name_plural", 51 "meta_model_name", 52 "flow_set", 53 ] 54 55 56class StageViewSet( 57 TypesMixin, 58 mixins.RetrieveModelMixin, 59 mixins.DestroyModelMixin, 60 UsedByMixin, 61 mixins.ListModelMixin, 62 GenericViewSet, 63): 64 """Stage Viewset""" 65 66 queryset = Stage.objects.none() 67 serializer_class = StageSerializer 68 search_fields = ["name"] 69 filterset_fields = ["name"] 70 71 def get_queryset(self): # pragma: no cover 72 return Stage.objects.select_subclasses().prefetch_related("flow_set") 73 74 @extend_schema(responses={200: UserSettingSerializer(many=True)}) 75 @action(detail=False, pagination_class=None, filter_backends=[]) 76 def user_settings(self, request: Request) -> Response: 77 """Get all stages the user can configure""" 78 stages = [] 79 for configurable_stage in all_subclasses(ConfigurableStage): 80 stages += list(configurable_stage.objects.all().order_by("name")) 81 matching_stages: list[dict] = [] 82 for stage in stages: 83 user_settings = stage.ui_user_settings() 84 if not user_settings: 85 continue 86 user_settings.initial_data["object_uid"] = str(stage.pk) 87 if hasattr(stage, "configure_flow") and stage.configure_flow: 88 user_settings.initial_data["configure_url"] = reverse( 89 "authentik_flows:configure", 90 kwargs={"stage_uuid": stage.pk}, 91 ) 92 if not user_settings.is_valid(): 93 LOGGER.warning(user_settings.errors) 94 matching_stages.append(user_settings.initial_data) 95 return Response(matching_stages)
LOGGER =
<BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
class
StageSerializer(authentik.core.api.utils.ModelSerializer, authentik.core.api.utils.MetaNameSerializer):
27class StageSerializer(ModelSerializer, MetaNameSerializer): 28 """Stage Serializer""" 29 30 component = SerializerMethodField() 31 flow_set = FlowSetSerializer(many=True, required=False, read_only=True) 32 33 def to_representation(self, instance: Stage): 34 if isinstance(instance, Stage) and instance.is_in_memory: 35 instance.stage_uuid = uuid4() 36 return super().to_representation(instance) 37 38 def get_component(self, obj: Stage) -> str: 39 """Get object type so that we know how to edit the object""" 40 if obj.__class__ == Stage: 41 return "" 42 return obj.component 43 44 class Meta: 45 model = Stage 46 fields = [ 47 "pk", 48 "name", 49 "component", 50 "verbose_name", 51 "verbose_name_plural", 52 "meta_model_name", 53 "flow_set", 54 ]
Stage Serializer
33 def to_representation(self, instance: Stage): 34 if isinstance(instance, Stage) and instance.is_in_memory: 35 instance.stage_uuid = uuid4() 36 return super().to_representation(instance)
Object instance -> Dict of primitive datatypes.
class
StageSerializer.Meta:
44 class Meta: 45 model = Stage 46 fields = [ 47 "pk", 48 "name", 49 "component", 50 "verbose_name", 51 "verbose_name_plural", 52 "meta_model_name", 53 "flow_set", 54 ]
model =
<class 'authentik.flows.models.Stage'>
class
StageViewSet(authentik.core.api.object_types.TypesMixin, rest_framework.mixins.RetrieveModelMixin, rest_framework.mixins.DestroyModelMixin, authentik.core.api.used_by.UsedByMixin, rest_framework.mixins.ListModelMixin, rest_framework.viewsets.GenericViewSet):
57class StageViewSet( 58 TypesMixin, 59 mixins.RetrieveModelMixin, 60 mixins.DestroyModelMixin, 61 UsedByMixin, 62 mixins.ListModelMixin, 63 GenericViewSet, 64): 65 """Stage Viewset""" 66 67 queryset = Stage.objects.none() 68 serializer_class = StageSerializer 69 search_fields = ["name"] 70 filterset_fields = ["name"] 71 72 def get_queryset(self): # pragma: no cover 73 return Stage.objects.select_subclasses().prefetch_related("flow_set") 74 75 @extend_schema(responses={200: UserSettingSerializer(many=True)}) 76 @action(detail=False, pagination_class=None, filter_backends=[]) 77 def user_settings(self, request: Request) -> Response: 78 """Get all stages the user can configure""" 79 stages = [] 80 for configurable_stage in all_subclasses(ConfigurableStage): 81 stages += list(configurable_stage.objects.all().order_by("name")) 82 matching_stages: list[dict] = [] 83 for stage in stages: 84 user_settings = stage.ui_user_settings() 85 if not user_settings: 86 continue 87 user_settings.initial_data["object_uid"] = str(stage.pk) 88 if hasattr(stage, "configure_flow") and stage.configure_flow: 89 user_settings.initial_data["configure_url"] = reverse( 90 "authentik_flows:configure", 91 kwargs={"stage_uuid": stage.pk}, 92 ) 93 if not user_settings.is_valid(): 94 LOGGER.warning(user_settings.errors) 95 matching_stages.append(user_settings.initial_data) 96 return Response(matching_stages)
Stage Viewset
serializer_class =
<class 'StageSerializer'>
def
get_queryset(self):
72 def get_queryset(self): # pragma: no cover 73 return Stage.objects.select_subclasses().prefetch_related("flow_set")
Get the list of items for this view.
This must be an iterable, and may be a queryset.
Defaults to using self.queryset.
This method should always be used rather than accessing self.queryset
directly, as self.queryset gets evaluated only once, and those results
are cached for all subsequent requests.
You may want to override this if you need to provide different querysets depending on the incoming request.
(Eg. return a list of items that is specific to the user)
@extend_schema(responses={200: UserSettingSerializer(many=True)})
@action(detail=False, pagination_class=None, filter_backends=[])
def
user_settings( self, request: rest_framework.request.Request) -> rest_framework.response.Response:
75 @extend_schema(responses={200: UserSettingSerializer(many=True)}) 76 @action(detail=False, pagination_class=None, filter_backends=[]) 77 def user_settings(self, request: Request) -> Response: 78 """Get all stages the user can configure""" 79 stages = [] 80 for configurable_stage in all_subclasses(ConfigurableStage): 81 stages += list(configurable_stage.objects.all().order_by("name")) 82 matching_stages: list[dict] = [] 83 for stage in stages: 84 user_settings = stage.ui_user_settings() 85 if not user_settings: 86 continue 87 user_settings.initial_data["object_uid"] = str(stage.pk) 88 if hasattr(stage, "configure_flow") and stage.configure_flow: 89 user_settings.initial_data["configure_url"] = reverse( 90 "authentik_flows:configure", 91 kwargs={"stage_uuid": stage.pk}, 92 ) 93 if not user_settings.is_valid(): 94 LOGGER.warning(user_settings.errors) 95 matching_stages.append(user_settings.initial_data) 96 return Response(matching_stages)
Get all stages the user can configure