authentik.enterprise.providers.ssf.views.base 

 1from django.http import Http404, HttpRequest
 2from rest_framework.permissions import IsAuthenticated
 3from rest_framework.views import APIView
 4from structlog.stdlib import BoundLogger, get_logger
 5
 6from authentik.core.models import Application
 7from authentik.enterprise.providers.ssf.models import SSFProvider, Stream, StreamStatus
 8from authentik.enterprise.providers.ssf.views.auth import SSFTokenAuth
 9
10
11class SSFView(APIView):
12    application: Application
13    provider: SSFProvider
14    logger: BoundLogger
15
16    permission_classes = [IsAuthenticated]
17
18    def setup(self, request: HttpRequest, *args, **kwargs) -> None:
19        self.logger = get_logger().bind()
20        super().setup(request, *args, **kwargs)
21
22    def get_authenticators(self):
23        return [SSFTokenAuth(self)]
24
25
26class SSFStreamView(SSFView):
27    def get_object(self, any_status=False) -> Stream:
28        streams = Stream.objects.filter(provider=self.provider)
29        if not any_status:
30            streams = streams.filter(status__in=[StreamStatus.ENABLED, StreamStatus.PAUSED])
31        if "stream_id" in self.request.query_params:
32            streams = streams.filter(pk=self.request.query_params["stream_id"])
33        if "stream_id" in self.request.data:
34            streams = streams.filter(pk=self.request.data["stream_id"])
35        stream = streams.first()
36        if not stream:
37            raise Http404()
38        return stream
class SSFView(rest_framework.views.APIView): 
12class SSFView(APIView):
13    application: Application
14    provider: SSFProvider
15    logger: BoundLogger
16
17    permission_classes = [IsAuthenticated]
18
19    def setup(self, request: HttpRequest, *args, **kwargs) -> None:
20        self.logger = get_logger().bind()
21        super().setup(request, *args, **kwargs)
22
23    def get_authenticators(self):
24        return [SSFTokenAuth(self)]

Intentionally simple parent class for all views. Only implements dispatch-by-method and simple sanity checking.

logger: structlog.stdlib.BoundLogger
permission_classes = [<class 'rest_framework.permissions.IsAuthenticated'>]
def setup(self, request: django.http.request.HttpRequest, *args, **kwargs) -> None: 
19    def setup(self, request: HttpRequest, *args, **kwargs) -> None:
20        self.logger = get_logger().bind()
21        super().setup(request, *args, **kwargs)

Initialize attributes shared by all view methods.

def get_authenticators(self): 
23    def get_authenticators(self):
24        return [SSFTokenAuth(self)]

Instantiates and returns the list of authenticators that this view can use.

class SSFStreamView(SSFView): 
27class SSFStreamView(SSFView):
28    def get_object(self, any_status=False) -> Stream:
29        streams = Stream.objects.filter(provider=self.provider)
30        if not any_status:
31            streams = streams.filter(status__in=[StreamStatus.ENABLED, StreamStatus.PAUSED])
32        if "stream_id" in self.request.query_params:
33            streams = streams.filter(pk=self.request.query_params["stream_id"])
34        if "stream_id" in self.request.data:
35            streams = streams.filter(pk=self.request.data["stream_id"])
36        stream = streams.first()
37        if not stream:
38            raise Http404()
39        return stream

Intentionally simple parent class for all views. Only implements dispatch-by-method and simple sanity checking.

def get_object( self, any_status=False) -> authentik.enterprise.providers.ssf.models.Stream: 
28    def get_object(self, any_status=False) -> Stream:
29        streams = Stream.objects.filter(provider=self.provider)
30        if not any_status:
31            streams = streams.filter(status__in=[StreamStatus.ENABLED, StreamStatus.PAUSED])
32        if "stream_id" in self.request.query_params:
33            streams = streams.filter(pk=self.request.query_params["stream_id"])
34        if "stream_id" in self.request.data:
35            streams = streams.filter(pk=self.request.data["stream_id"])
36        stream = streams.first()
37        if not stream:
38            raise Http404()
39        return stream
Inherited Members
SSFView
application
provider
logger
permission_classes
setup
get_authenticators