Skip to content

characteristics_view#

FileCharacteristicsQueryParamsSerializer#

Bases: Serializer

Source code in src/apps/files/views/characteristics_view.py 
16
17
18
19
20
class FileCharacteristicsQueryParamsSerializer(serializers.Serializer):
    dataset = serializers.UUIDField(
        default=None,
        help_text="Use to access characteristics as a member of a dataset the file belongs to.",
    )

FileCharacteristicsViewSet#

Bases: CommonModelViewSet

Source code in src/apps/files/views/characteristics_view.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class FileCharacteristicsViewSet(CommonModelViewSet):
    http_method_names = ["get", "put", "patch", "delete", "options"]
    access_policy = FileCharacteristicsAccessPolicy
    serializer_class = FileCharacteristicsSerializer
    query_serializers = [{"class": FileCharacteristicsQueryParamsSerializer}]

    def get_queryset(self):
        if getattr(
            self, "swagger_fake_view", None
        ):  # kwargs are not available in swagger inspection
            return FileCharacteristics.objects.none()

        file = self.get_file_instance()
        return FileCharacteristics.objects.filter(file=file)

    def sync(self, file: File):
        sync_files.send(
            sender=File,
            actions=[{"action": BulkAction.UPDATE, "object": file}],
        )

    def perform_create(self, serializer):
        file = self.get_file_instance()
        characteristics = serializer.save()
        File.objects.filter(id=file.id).update(characteristics=characteristics)
        self.sync(file)

    def update(self, request, *args, **kwargs):
        if not kwargs.get("patch"):
            # Allow creating new object when in PUT mode
            if not self.get_queryset().exists():
                return self.create(request, *args, **kwargs)

        resp = super().update(request, *args, **kwargs)
        self.sync(self.get_file_instance())
        return resp

    def destroy(self, request, *args, **kwargs):
        resp = super().destroy(request, *args, **kwargs)
        self.sync(self.get_file_instance())
        return resp

    def get_serializer_context(self):
        context = super().get_serializer_context()
        if getattr(self, "swagger_fake_view", None):
            # kwargs are not available in swagger inspection
            return context
        context["file"] = self.get_file_instance()
        return context

    def get_file_instance(self) -> File:
        if instance := getattr(self, "_file_instance", None):
            return instance
        queryset = File.available_objects.filter(id=self.kwargs["pk"])
        queryset = FilesAccessPolicy().scope_queryset(
            self.request, queryset, dataset_id=self.query_params["dataset"]
        )
        self._file_instance = get_object_or_404(queryset)
        return self._file_instance

    def get_object(self):
        queryset = self.get_queryset()
        obj = get_object_or_404(queryset)

        # May raise a permission denied
        self.check_object_permissions(self.request, obj)
        return obj