authentik.enterprise.reports.models

View Source

  1import csv
  2import io
  3from uuid import uuid4
  4
  5from django.contrib.contenttypes.models import ContentType
  6from django.db import models
  7from django.utils.translation import gettext as _
  8from rest_framework.serializers import Serializer
  9from rest_framework.viewsets import ModelViewSet
 10
 11from authentik.admin.files.fields import FileField
 12from authentik.admin.files.manager import get_file_manager
 13from authentik.admin.files.usage import FileUsage
 14from authentik.core.models import User
 15from authentik.enterprise.reports.utils import MockRequest
 16from authentik.events.models import Event, EventAction, Notification, NotificationSeverity
 17from authentik.lib.models import SerializerModel
 18from authentik.lib.utils.db import chunked_queryset
 19from authentik.tenants.utils import get_current_tenant
 20
 21
 22class DataExport(SerializerModel):
 23    id = models.UUIDField(primary_key=True, default=uuid4)
 24    requested_by = models.ForeignKey(User, null=True, on_delete=models.SET_NULL)
 25    requested_on = models.DateTimeField(auto_now_add=True)
 26    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
 27    query_params = models.JSONField()
 28    file = FileField(blank=True)
 29    completed = models.BooleanField(default=False)
 30
 31    class Meta:
 32        verbose_name = _("Data Export")
 33        verbose_name_plural = _("Data Exports")
 34
 35    @property
 36    def serializer(self) -> type[Serializer]:
 37        """Get serializer for this model"""
 38        from authentik.enterprise.reports.api.reports import DataExportSerializer
 39
 40        return DataExportSerializer
 41
 42    def generate(self) -> None:
 43        if self.completed:
 44            raise AssertionError("Data export must only be generated once")
 45
 46        model_class = self.content_type.model_class()
 47        model_verbose_name = model_class._meta.verbose_name
 48        model_verbose_name_plural = model_class._meta.verbose_name_plural
 49
 50        queryset = chunked_queryset(self.get_queryset())
 51
 52        serializer = self.get_serializer_class()(
 53            context={"request": self._get_request()}, instance=queryset, many=True
 54        )
 55        self.file = f"{model_verbose_name_plural.lower()}_{self.id}.csv"
 56
 57        with get_file_manager(FileUsage.REPORTS).save_file_stream(self.file) as f:
 58            with io.TextIOWrapper(f, encoding="utf-8", newline="") as text:
 59                writer = csv.writer(text)
 60                fields = [field.label for field in serializer.child.fields.values()]
 61                writer.writerow(fields)
 62                for record in queryset:
 63                    data = serializer.child.to_representation(record).values()
 64                    writer.writerow(data)
 65        self.completed = True
 66        self.save()
 67
 68        message = _(f"{model_verbose_name} export generated successfully")
 69        e = Event.new(
 70            EventAction.EXPORT_READY,
 71            message=message,
 72            export=self,
 73        ).set_user(self.requested_by)
 74        e.save()
 75        Notification.objects.create(
 76            event=e,
 77            severity=NotificationSeverity.NOTICE,
 78            body=message,
 79            hyperlink=self.file_url,
 80            hyperlink_label=_("Download"),
 81            user=self.requested_by,
 82        )
 83
 84    @property
 85    def file_url(self) -> str:
 86        return get_file_manager(FileUsage.REPORTS).file_url(self.file)
 87
 88    def _get_request(self) -> MockRequest:
 89        return MockRequest(
 90            user=self.requested_by, query_params=self.query_params, tenant=get_current_tenant()
 91        )
 92
 93    def get_queryset(self) -> models.QuerySet:
 94        request = self._get_request()
 95        viewset = self.get_viewset()
 96        viewset.request = request
 97        queryset = viewset.get_queryset()
 98        queryset = viewset.filter_queryset(queryset)
 99
100        return queryset
101
102    def get_viewset(self) -> ModelViewSet:
103        from authentik.core.api.users import UserViewSet
104        from authentik.events.api.events import EventViewSet
105
106        model = (self.content_type.app_label, self.content_type.model)
107        if model == ("authentik_core", "user"):
108            return UserViewSet()
109        elif model == ("authentik_events", "event"):
110            return EventViewSet()
111        raise NotImplementedError(f"Unsupported data export type {self.content_type.model}")
112
113    def get_serializer_class(self) -> type[Serializer]:
114        from authentik.enterprise.reports.serializers import (
115            ExportEventSerializer,
116            ExportUserSerializer,
117        )
118
119        if self.content_type.model == "user":
120            return ExportUserSerializer
121        elif self.content_type.model == "event":
122            return ExportEventSerializer
123        return self.get_viewset().get_serializer_class()

class DataExport(authentik.lib.models.SerializerModel): View Source

 23class DataExport(SerializerModel):
 24    id = models.UUIDField(primary_key=True, default=uuid4)
 25    requested_by = models.ForeignKey(User, null=True, on_delete=models.SET_NULL)
 26    requested_on = models.DateTimeField(auto_now_add=True)
 27    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
 28    query_params = models.JSONField()
 29    file = FileField(blank=True)
 30    completed = models.BooleanField(default=False)
 31
 32    class Meta:
 33        verbose_name = _("Data Export")
 34        verbose_name_plural = _("Data Exports")
 35
 36    @property
 37    def serializer(self) -> type[Serializer]:
 38        """Get serializer for this model"""
 39        from authentik.enterprise.reports.api.reports import DataExportSerializer
 40
 41        return DataExportSerializer
 42
 43    def generate(self) -> None:
 44        if self.completed:
 45            raise AssertionError("Data export must only be generated once")
 46
 47        model_class = self.content_type.model_class()
 48        model_verbose_name = model_class._meta.verbose_name
 49        model_verbose_name_plural = model_class._meta.verbose_name_plural
 50
 51        queryset = chunked_queryset(self.get_queryset())
 52
 53        serializer = self.get_serializer_class()(
 54            context={"request": self._get_request()}, instance=queryset, many=True
 55        )
 56        self.file = f"{model_verbose_name_plural.lower()}_{self.id}.csv"
 57
 58        with get_file_manager(FileUsage.REPORTS).save_file_stream(self.file) as f:
 59            with io.TextIOWrapper(f, encoding="utf-8", newline="") as text:
 60                writer = csv.writer(text)
 61                fields = [field.label for field in serializer.child.fields.values()]
 62                writer.writerow(fields)
 63                for record in queryset:
 64                    data = serializer.child.to_representation(record).values()
 65                    writer.writerow(data)
 66        self.completed = True
 67        self.save()
 68
 69        message = _(f"{model_verbose_name} export generated successfully")
 70        e = Event.new(
 71            EventAction.EXPORT_READY,
 72            message=message,
 73            export=self,
 74        ).set_user(self.requested_by)
 75        e.save()
 76        Notification.objects.create(
 77            event=e,
 78            severity=NotificationSeverity.NOTICE,
 79            body=message,
 80            hyperlink=self.file_url,
 81            hyperlink_label=_("Download"),
 82            user=self.requested_by,
 83        )
 84
 85    @property
 86    def file_url(self) -> str:
 87        return get_file_manager(FileUsage.REPORTS).file_url(self.file)
 88
 89    def _get_request(self) -> MockRequest:
 90        return MockRequest(
 91            user=self.requested_by, query_params=self.query_params, tenant=get_current_tenant()
 92        )
 93
 94    def get_queryset(self) -> models.QuerySet:
 95        request = self._get_request()
 96        viewset = self.get_viewset()
 97        viewset.request = request
 98        queryset = viewset.get_queryset()
 99        queryset = viewset.filter_queryset(queryset)
100
101        return queryset
102
103    def get_viewset(self) -> ModelViewSet:
104        from authentik.core.api.users import UserViewSet
105        from authentik.events.api.events import EventViewSet
106
107        model = (self.content_type.app_label, self.content_type.model)
108        if model == ("authentik_core", "user"):
109            return UserViewSet()
110        elif model == ("authentik_events", "event"):
111            return EventViewSet()
112        raise NotImplementedError(f"Unsupported data export type {self.content_type.model}")
113
114    def get_serializer_class(self) -> type[Serializer]:
115        from authentik.enterprise.reports.serializers import (
116            ExportEventSerializer,
117            ExportUserSerializer,
118        )
119
120        if self.content_type.model == "user":
121            return ExportUserSerializer
122        elif self.content_type.model == "event":
123            return ExportEventSerializer
124        return self.get_viewset().get_serializer_class()

DataExport(id, requested_by, requested_on, content_type, query_params, file, completed)

def id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

requested_by

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def requested_on(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

content_type

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

def query_params(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def file(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def completed(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

serializer: type[rest_framework.serializers.Serializer] View Source

36    @property
37    def serializer(self) -> type[Serializer]:
38        """Get serializer for this model"""
39        from authentik.enterprise.reports.api.reports import DataExportSerializer
40
41        return DataExportSerializer

Get serializer for this model

def generate(self) -> None: View Source

43    def generate(self) -> None:
44        if self.completed:
45            raise AssertionError("Data export must only be generated once")
46
47        model_class = self.content_type.model_class()
48        model_verbose_name = model_class._meta.verbose_name
49        model_verbose_name_plural = model_class._meta.verbose_name_plural
50
51        queryset = chunked_queryset(self.get_queryset())
52
53        serializer = self.get_serializer_class()(
54            context={"request": self._get_request()}, instance=queryset, many=True
55        )
56        self.file = f"{model_verbose_name_plural.lower()}_{self.id}.csv"
57
58        with get_file_manager(FileUsage.REPORTS).save_file_stream(self.file) as f:
59            with io.TextIOWrapper(f, encoding="utf-8", newline="") as text:
60                writer = csv.writer(text)
61                fields = [field.label for field in serializer.child.fields.values()]
62                writer.writerow(fields)
63                for record in queryset:
64                    data = serializer.child.to_representation(record).values()
65                    writer.writerow(data)
66        self.completed = True
67        self.save()
68
69        message = _(f"{model_verbose_name} export generated successfully")
70        e = Event.new(
71            EventAction.EXPORT_READY,
72            message=message,
73            export=self,
74        ).set_user(self.requested_by)
75        e.save()
76        Notification.objects.create(
77            event=e,
78            severity=NotificationSeverity.NOTICE,
79            body=message,
80            hyperlink=self.file_url,
81            hyperlink_label=_("Download"),
82            user=self.requested_by,
83        )

file_url: str View Source

85    @property
86    def file_url(self) -> str:
87        return get_file_manager(FileUsage.REPORTS).file_url(self.file)

def get_queryset(self) -> django.db.models.query.QuerySet: View Source

 94    def get_queryset(self) -> models.QuerySet:
 95        request = self._get_request()
 96        viewset = self.get_viewset()
 97        viewset.request = request
 98        queryset = viewset.get_queryset()
 99        queryset = viewset.filter_queryset(queryset)
100
101        return queryset

def get_viewset(self) -> rest_framework.viewsets.ModelViewSet: View Source

103    def get_viewset(self) -> ModelViewSet:
104        from authentik.core.api.users import UserViewSet
105        from authentik.events.api.events import EventViewSet
106
107        model = (self.content_type.app_label, self.content_type.model)
108        if model == ("authentik_core", "user"):
109            return UserViewSet()
110        elif model == ("authentik_events", "event"):
111            return EventViewSet()
112        raise NotImplementedError(f"Unsupported data export type {self.content_type.model}")

def get_serializer_class(self) -> type[rest_framework.serializers.Serializer]: View Source

114    def get_serializer_class(self) -> type[Serializer]:
115        from authentik.enterprise.reports.serializers import (
116            ExportEventSerializer,
117            ExportUserSerializer,
118        )
119
120        if self.content_type.model == "user":
121            return ExportUserSerializer
122        elif self.content_type.model == "event":
123            return ExportEventSerializer
124        return self.get_viewset().get_serializer_class()

requested_by_id

def get_next_by_requested_on(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

def get_previous_by_requested_on(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

content_type_id

def objects(unknown):

The type of the None singleton.

Inherited Members

authentik.lib.models.SerializerModel: Meta

class DataExport.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class DataExport.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.