authentik.tasks.api.workers 

 1from django.utils.timezone import now, timedelta
 2from drf_spectacular.utils import extend_schema, inline_serializer
 3from packaging.version import parse
 4from rest_framework.fields import BooleanField, CharField
 5from rest_framework.request import Request
 6from rest_framework.response import Response
 7from rest_framework.views import APIView
 8
 9from authentik import authentik_full_version
10from authentik.rbac.permissions import HasPermission
11from authentik.tasks.models import WorkerStatus
12
13
14class WorkerView(APIView):
15    """Get currently connected worker count."""
16
17    permission_classes = [HasPermission("authentik_rbac.view_system_info")]
18
19    @extend_schema(
20        responses=inline_serializer(
21            "Worker",
22            fields={
23                "worker_id": CharField(),
24                "version": CharField(),
25                "version_matching": BooleanField(),
26            },
27            many=True,
28        )
29    )
30    def get(self, request: Request) -> Response:
31        response = []
32        our_version = parse(authentik_full_version())
33        for status in WorkerStatus.objects.filter(last_seen__gt=now() - timedelta(seconds=45)):
34            version_matching = parse(status.version) == our_version
35            response.append(
36                {
37                    "worker_id": f"{status.pk}@{status.hostname}",
38                    "version": status.version,
39                    "version_matching": version_matching,
40                }
41            )
42        return Response(response)
class WorkerView(rest_framework.views.APIView): 
15class WorkerView(APIView):
16    """Get currently connected worker count."""
17
18    permission_classes = [HasPermission("authentik_rbac.view_system_info")]
19
20    @extend_schema(
21        responses=inline_serializer(
22            "Worker",
23            fields={
24                "worker_id": CharField(),
25                "version": CharField(),
26                "version_matching": BooleanField(),
27            },
28            many=True,
29        )
30    )
31    def get(self, request: Request) -> Response:
32        response = []
33        our_version = parse(authentik_full_version())
34        for status in WorkerStatus.objects.filter(last_seen__gt=now() - timedelta(seconds=45)):
35            version_matching = parse(status.version) == our_version
36            response.append(
37                {
38                    "worker_id": f"{status.pk}@{status.hostname}",
39                    "version": status.version,
40                    "version_matching": version_matching,
41                }
42            )
43        return Response(response)

Get currently connected worker count.

permission_classes = [<class 'authentik.rbac.permissions.HasPermission.<locals>.checker'>]
@extend_schema(responses=inline_serializer('Worker', fields={'worker_id': CharField(), 'version': CharField(), 'version_matching': BooleanField()}, many=True))
def get( self, request: rest_framework.request.Request) -> rest_framework.response.Response: 
20    @extend_schema(
21        responses=inline_serializer(
22            "Worker",
23            fields={
24                "worker_id": CharField(),
25                "version": CharField(),
26                "version_matching": BooleanField(),
27            },
28            many=True,
29        )
30    )
31    def get(self, request: Request) -> Response:
32        response = []
33        our_version = parse(authentik_full_version())
34        for status in WorkerStatus.objects.filter(last_seen__gt=now() - timedelta(seconds=45)):
35            version_matching = parse(status.version) == our_version
36            response.append(
37                {
38                    "worker_id": f"{status.pk}@{status.hostname}",
39                    "version": status.version,
40                    "version_matching": version_matching,
41                }
42            )
43        return Response(response)