This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/admin/api/overview.py

81 lines
2.8 KiB
Python

"""passbook administration overview"""
from django.core.cache import cache
from django.http import response
from drf_yasg2.utils import swagger_auto_schema
from rest_framework.fields import SerializerMethodField
from rest_framework.permissions import IsAdminUser
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.serializers import Serializer
from rest_framework.viewsets import ViewSet
from passbook import __version__
from passbook.admin.tasks import VERSION_CACHE_KEY, update_latest_version
from passbook.core.models import Provider
from passbook.policies.models import Policy
from passbook.root.celery import CELERY_APP
class AdministrationOverviewSerializer(Serializer):
"""Overview View"""
version = SerializerMethodField()
version_latest = SerializerMethodField()
worker_count = SerializerMethodField()
providers_without_application = SerializerMethodField()
policies_without_binding = SerializerMethodField()
cached_policies = SerializerMethodField()
cached_flows = SerializerMethodField()
def get_version(self, _) -> str:
"""Get current version"""
return __version__
def get_version_latest(self, _) -> str:
"""Get latest version from cache"""
version_in_cache = cache.get(VERSION_CACHE_KEY)
if not version_in_cache:
update_latest_version.delay()
return __version__
return version_in_cache
def get_worker_count(self, _) -> int:
"""Ping workers"""
return len(CELERY_APP.control.ping(timeout=0.5))
def get_providers_without_application(self, _) -> int:
"""Count of providers without application"""
return len(Provider.objects.filter(application=None))
def get_policies_without_binding(self, _) -> int:
"""Count of policies not bound or use in prompt stages"""
return len(
Policy.objects.filter(bindings__isnull=True, promptstage__isnull=True)
)
def get_cached_policies(self, _) -> int:
"""Get cached policy count"""
return len(cache.keys("policy_*"))
def get_cached_flows(self, _) -> int:
"""Get cached flow count"""
return len(cache.keys("flow_*"))
def create(self, request: Request) -> response:
raise NotImplementedError
def update(self, request: Request) -> Response:
raise NotImplementedError
class AdministrationOverviewViewSet(ViewSet):
"""Return single instance of AdministrationOverviewSerializer"""
permission_classes = [IsAdminUser]
@swagger_auto_schema(responses={200: AdministrationOverviewSerializer(many=True)})
def list(self, request: Request) -> Response:
"""Return single instance of AdministrationOverviewSerializer"""
serializer = AdministrationOverviewSerializer(True)
return Response(serializer.data)