Skip to content

nested_views#

DatasetNestedViewSet#

Bases: CommonModelViewSet

Source code in src/apps/core/views/nested_views.py 
19
20
21
22
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
class DatasetNestedViewSet(CommonModelViewSet):
    access_policy = DatasetNestedAccessPolicy

    def get_queryset(self):
        if getattr(
            self, "swagger_fake_view", None
        ):  # kwargs are not available in swagger inspection
            return self.serializer_class.Meta.model.available_objects.none()

        dataset = self.get_dataset_instance()
        return self.serializer_class.Meta.model.available_objects.filter(dataset=dataset)

    def create(self, request, *args, **kwargs):
        resp = super().create(request, *args, **kwargs)
        self.get_dataset_instance().signal_update()
        return resp

    def update(self, request, *args, **kwargs):
        resp = super().update(request, *args, **kwargs)
        self.get_dataset_instance().signal_update()
        return resp

    def perform_create(self, serializer):
        dataset = self.get_dataset_instance()
        return serializer.save(dataset_id=dataset.id)

    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["dataset"] = self.get_dataset_instance()
        return context

    def get_dataset_instance(self) -> Dataset:
        dataset_qs = Dataset.available_objects.filter(id=self.kwargs["dataset_pk"])
        dataset_qs = DatasetAccessPolicy().scope_queryset(self.request, dataset_qs)
        return get_object_or_404(dataset_qs)

DatasetNestedOneToOneView#

Bases: DatasetNestedViewSet

Source code in src/apps/core/views/nested_views.py 
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
90
91
class DatasetNestedOneToOneView(DatasetNestedViewSet):
    http_method_names = ["get", "put", "patch", "delete", "options"]

    def __init__(self, *args, **kwargs):
        if not getattr(self, "dataset_field_name", None):
            raise ValueError(f"Missing dataset_field_name from {self.__class__.__name__}")
        super().__init__(*args, **kwargs)

    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

    def update(self, request, *args, **kwargs):
        obj = None
        if not kwargs.get("patch"):
            # Allow creating new object when in PUT mode
            queryset = self.get_queryset()
            (
                obj,
                created,
            ) = queryset.get_or_create()  # Create initial object if it does not exist yet
            if created:
                from apps.core.models.catalog_record.dataset import Dataset

                # Assign created object to Dataset.<dataset_field_name>
                Dataset.objects.filter(id=self.kwargs["dataset_pk"]).update(
                    **{self.dataset_field_name: obj}
                )
        return super().update(request, *args, **kwargs)