Skip to content

file_view#

FileCommonFilterset#

Bases: FilterSet

File attribute specific filters for files.

Source code in src/apps/files/views/file_view.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class FileCommonFilterset(filters.FilterSet):
    """File attribute specific filters for files."""

    filename = filters.CharFilter(
        lookup_expr="icontains",
    )
    pathname = filters.CharFilter(method="pathname_filter")

    storage_identifier = filters.CharFilter()

    frozen__gt = filters.DateTimeFilter(field_name="frozen", lookup_expr="gt")

    def pathname_filter(self, queryset, name, value):
        if value.endswith("/"):
            # Filtering by directory path, no need to include filename
            return queryset.filter(directory_path__istartswith=value)
        return queryset.alias(pathname=Concat("directory_path", "filename")).filter(
            pathname__istartswith=value
        )

    class Meta:
        model = File
        fields = ()

FileFilterSet#

Bases: FileCommonFilterset

Add project and dataset filters to file filterset.

Source code in src/apps/files/views/file_view.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class FileFilterSet(FileCommonFilterset):
    """Add project and dataset filters to file filterset."""

    csc_project = filters.CharFilter(
        field_name="storage__csc_project",
        max_length=200,
    )
    storage_service = VerboseChoiceFilter(
        field_name="storage__storage_service",
        choices=[(v, v) for v in settings.STORAGE_SERVICE_FILE_STORAGES],
    )

    dataset = filters.UUIDFilter(field_name="file_sets__dataset_id")

    class Meta:
        model = File
        fields = ()

FileDeleteListFilterSet#

Bases: FileFilterSet

Add project and dataset filters to file filterset.

Source code in src/apps/files/views/file_view.py 
85
86
87
88
89
90
class FileDeleteListFilterSet(FileFilterSet):
    """Add project and dataset filters to file filterset."""

    csc_project = filters.CharFilter(
        field_name="storage__csc_project", max_length=200, required=True
    )

FilesDatasetsQueryParamsSerializer#

Bases: Serializer

Source code in src/apps/files/views/file_view.py 
93
94
95
96
97
class FilesDatasetsQueryParamsSerializer(serializers.Serializer):
    relations = serializers.BooleanField(
        default=False, help_text="List dataset relations per file"
    )
    storage_service = StorageServiceField(default=None)

FilesDatasetsBodySerializer#

Bases: ListSerializer

Source code in src/apps/files/views/file_view.py 
100
101
class FilesDatasetsBodySerializer(serializers.ListSerializer):
    child = serializers.CharField()

BaseFileViewSet#

Bases: CommonModelViewSet

Basic read-only files view.

Source code in src/apps/files/views/file_view.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class BaseFileViewSet(CommonModelViewSet):
    """Basic read-only files view."""

    serializer_class = FileSerializer
    filterset_class = FileFilterSet
    http_method_names = ["get"]
    queryset = File.available_objects.prefetch_related("storage")
    queryset_include_removed = File.all_objects.prefetch_related("storage")
    access_policy: FilesAccessPolicy = FilesAccessPolicy

    # Query serializer info for query_params and swagger generation
    query_serializers = [
        {
            "class": IncludeRemovedQueryParamsSerializer,
            "actions": ["list", "retrieve"],
        }
    ]

    def get_queryset(self):
        queryset = self.queryset
        if self.query_params.get("include_removed"):
            queryset = self.queryset_include_removed

        return self.access_policy.scope_queryset(
            self.request, queryset, dataset_id=self.get_dataset_id()
        )

    def retrieve(self, request, *args, **kwargs):
        return super().retrieve(request, *args, **kwargs)

    def list(self, request, *args, **kwargs):
        pagination_enabled = self.paginator.pagination_enabled(request)
        with cachalot_toggle(pagination_enabled):
            return super().list(request, *args, **kwargs)

    def get_serializer(self, instance=None, *args, **kwargs):
        """Modified get_serializer that passes instance to get_serializer_context."""
        serializer_class = self.get_serializer_class()
        kwargs.setdefault("context", self.get_serializer_context(instance))
        return serializer_class(instance, *args, **kwargs)

    def get_dataset_id(self):
        """Get dataset id from kwargs (i.e. from url) or query parameters."""
        dataset_id = self.kwargs.get("dataset_id")
        if not dataset_id and self.request:
            dataset_id = forms.CharField(required=False).clean(
                self.request.query_params.get("dataset")
            )
        return dataset_id

    def get_serializer_context(self, instance):
        """Add dataset file metadata to serializer context when listing files."""
        context = super().get_serializer_context()
        if self.request and self.request.method != "GET":
            return context

        if dataset_id := self.get_dataset_id():
            # Convert single instance to list
            files = instance
            if not (isinstance(files, list) or isinstance(files, QuerySet)):
                files = [files]

            # Get file metadata objects as dict by file id
            file_metadata = (
                get_file_metadata_model()
                .objects.filter(file_set__dataset_id=dataset_id)
                .prefetch_related("file_type")
                .distinct("file_id")
                .in_bulk([f.id for f in files], field_name="file_id")
            )
            context["file_metadata"] = file_metadata
        return context

get_dataset_id() #

Get dataset id from kwargs (i.e. from url) or query parameters.

Source code in src/apps/files/views/file_view.py 
145
146
147
148
149
150
151
152
def get_dataset_id(self):
    """Get dataset id from kwargs (i.e. from url) or query parameters."""
    dataset_id = self.kwargs.get("dataset_id")
    if not dataset_id and self.request:
        dataset_id = forms.CharField(required=False).clean(
            self.request.query_params.get("dataset")
        )
    return dataset_id

get_serializer(instance=None, *args, **kwargs) #

Modified get_serializer that passes instance to get_serializer_context.

Source code in src/apps/files/views/file_view.py 
139
140
141
142
143
def get_serializer(self, instance=None, *args, **kwargs):
    """Modified get_serializer that passes instance to get_serializer_context."""
    serializer_class = self.get_serializer_class()
    kwargs.setdefault("context", self.get_serializer_context(instance))
    return serializer_class(instance, *args, **kwargs)

get_serializer_context(instance) #

Add dataset file metadata to serializer context when listing files.

Source code in src/apps/files/views/file_view.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def get_serializer_context(self, instance):
    """Add dataset file metadata to serializer context when listing files."""
    context = super().get_serializer_context()
    if self.request and self.request.method != "GET":
        return context

    if dataset_id := self.get_dataset_id():
        # Convert single instance to list
        files = instance
        if not (isinstance(files, list) or isinstance(files, QuerySet)):
            files = [files]

        # Get file metadata objects as dict by file id
        file_metadata = (
            get_file_metadata_model()
            .objects.filter(file_set__dataset_id=dataset_id)
            .prefetch_related("file_type")
            .distinct("file_id")
            .in_bulk([f.id for f in files], field_name="file_id")
        )
        context["file_metadata"] = file_metadata
    return context

FileBulkQuerySerializer#

Bases: Serializer

Source code in src/apps/files/views/file_view.py 
178
179
180
181
class FileBulkQuerySerializer(serializers.Serializer):
    ignore_errors = serializers.BooleanField(
        default=False, help_text=_("Commit changes and return 200 even if there are errors.")
    )

FileViewSet#

Bases: BaseFileViewSet

Source code in src/apps/files/views/file_view.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class FileViewSet(BaseFileViewSet):
    http_method_names = ["get", "post", "patch", "put", "delete"]

    # TODO: Restore files action (=convert removed files to "not removed", should not undeprecate datasets)

    query_serializers = BaseFileViewSet.query_serializers + [
        {
            "class": FilesDatasetsQueryParamsSerializer,
            "actions": ["datasets"],
        },
        {
            "class": FileBulkQuerySerializer,
            "actions": ["post_many", "patch_many", "put_many", "delete_many"],
        },
        {"class": FlushQueryParamsSerializer, "actions": ["destroy_list"]},
    ]

    @property
    def filterset_class(self):
        if self.action == "destroy_list":
            # DELETE on list requires project to be set
            return FileDeleteListFilterSet
        else:
            return FileFilterSet

    @swagger_auto_schema(
        request_body=FilesDatasetsBodySerializer,
    )
    @action(detail=False, methods=["post"])
    def datasets(self, request):
        """Return datasets belonging to files.

        The request body should contain an array of file identifiers.
        By default, the file identifiers are internal Metax `id` values.
        If `storage_service` is set, the identifiers are `storage_identifier`
        values specific to that storage service. The file identifiers in
        the response will be of the same type as the input identifiers.

        `relations=false` (default): Return list of dataset ids.

        `relations=true`: Return object with a list of dataset ids
        for each file identifier. Files with no datasets are omitted.

        `storage_service`: List only files in specific storage service, and
        use `storage_identifier` as file identifier in both input and output.


        POST is used instead of GET because of query parameter length limitations for GET requests.
        """

        body_serializer = FilesDatasetsBodySerializer(data=self.request.data)
        body_serializer.is_valid(raise_exception=True)
        ids = body_serializer.validated_data

        # File identifiers are metax internal ids unless storage_service is defined
        file_id_type = "id"

        # Allow limiting results to specific storage_service
        params = self.query_params
        storage_filter = Q()
        if storage_service := params["storage_service"]:
            file_id_type = "storage_identifier"
            storage_filter = Q(storage__storage_service=storage_service)

        try:
            if params["relations"]:
                # Return dict of file ids -> list of dataset ids
                file_id_filter = Q(**{f"{file_id_type}__in": ids})
                queryset = File.objects.filter(
                    file_id_filter, storage_filter, file_sets__isnull=False
                ).values(key=F(file_id_type))
                queryset = queryset.annotate(
                    values=ArrayAgg(
                        "file_sets__dataset_id",
                        filter=Q(file_sets__dataset__deprecated__isnull=True),
                        default=Value([]),
                    )
                )
                return Response({str(v["key"]): v["values"] for v in queryset})
            else:
                # Return list of dataset ids
                file_id_filter = Q(**{f"files__{file_id_type}__in": ids})
                file_set_model = File.file_sets.rel.related_model
                queryset = (
                    file_set_model.objects.filter(file_id_filter, storage_filter)
                    .values_list("dataset__id", flat=True)
                    .distinct()
                )
                return Response(queryset)

        except DjangoValidationError as e:
            raise serializers.ValidationError(e.messages)  # avoid 500 from invalid uuid

    def bulk_action(self, files, action):
        ignore_errors = self.query_params["ignore_errors"]
        serializer = FileBulkSerializer(data=files, action=action, ignore_errors=ignore_errors)
        serializer.is_valid(raise_exception=True)
        serializer.save()

        status = 200
        if serializer.data.get("failed") and not ignore_errors:
            status = 400
        return Response(serializer.data, status=status)

    @swagger_auto_schema(
        request_body=FileBulkSerializer(action=BulkAction.INSERT),
        responses={200: FileBulkReturnValueSerializer()},
    )
    @action(detail=False, methods=["post"], url_path="post-many")
    def post_many(self, request):
        return self.bulk_action(request.data, action=BulkAction.INSERT)

    @swagger_auto_schema(
        request_body=FileBulkSerializer(action=BulkAction.UPDATE),
        responses={200: FileBulkReturnValueSerializer()},
    )
    @action(detail=False, methods=["post"], url_path="patch-many")
    def patch_many(self, request):
        return self.bulk_action(request.data, action=BulkAction.UPDATE)

    @swagger_auto_schema(
        request_body=FileBulkSerializer(action=BulkAction.UPSERT),
        responses={200: FileBulkReturnValueSerializer()},
    )
    @action(detail=False, methods=["post"], url_path="put-many")
    def put_many(self, request):
        return self.bulk_action(request.data, action=BulkAction.UPSERT)

    @swagger_auto_schema(
        request_body=FileBulkSerializer(action=BulkAction.DELETE),
        responses={200: FileBulkReturnValueSerializer()},
    )
    @action(detail=False, methods=["post"], url_path="delete-many")
    def delete_many(self, request):
        return self.bulk_action(request.data, action=BulkAction.DELETE)

    @swagger_auto_schema(
        operation_id="v3_files_delete_list",
        manual_parameters=get_filter_openapi_parameters(FileDeleteListFilterSet),
        responses={200: DeleteListReturnValueSerializer()},
    )
    def destroy_list(self, *args, **kwargs):
        """Delete files matching query parameters.

        The `csc_project` parameter is required. If no `storage_service` is defined,
        matching files from all storage services are deleted.

        By default the files are flagged as removed.
        When flush is enabled, the files are removed from the database.
        """

        flush = self.query_params["flush"]
        queryset: QuerySet
        if flush:
            queryset = File.all_objects
        else:
            queryset = File.available_objects
        queryset = self.filter_queryset(queryset)

        count = queryset.count()
        queryset.delete()

        return Response(DeleteListReturnValueSerializer(instance={"count": count}).data, 200)

datasets(request) #

Return datasets belonging to files.

The request body should contain an array of file identifiers. By default, the file identifiers are internal Metax id values. If storage_service is set, the identifiers are storage_identifier values specific to that storage service. The file identifiers in the response will be of the same type as the input identifiers.

relations=false (default): Return list of dataset ids.

relations=true: Return object with a list of dataset ids for each file identifier. Files with no datasets are omitted.

storage_service: List only files in specific storage service, and use storage_identifier as file identifier in both input and output.

POST is used instead of GET because of query parameter length limitations for GET requests.

Source code in src/apps/files/views/file_view.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@swagger_auto_schema(
    request_body=FilesDatasetsBodySerializer,
)
@action(detail=False, methods=["post"])
def datasets(self, request):
    """Return datasets belonging to files.

    The request body should contain an array of file identifiers.
    By default, the file identifiers are internal Metax `id` values.
    If `storage_service` is set, the identifiers are `storage_identifier`
    values specific to that storage service. The file identifiers in
    the response will be of the same type as the input identifiers.

    `relations=false` (default): Return list of dataset ids.

    `relations=true`: Return object with a list of dataset ids
    for each file identifier. Files with no datasets are omitted.

    `storage_service`: List only files in specific storage service, and
    use `storage_identifier` as file identifier in both input and output.


    POST is used instead of GET because of query parameter length limitations for GET requests.
    """

    body_serializer = FilesDatasetsBodySerializer(data=self.request.data)
    body_serializer.is_valid(raise_exception=True)
    ids = body_serializer.validated_data

    # File identifiers are metax internal ids unless storage_service is defined
    file_id_type = "id"

    # Allow limiting results to specific storage_service
    params = self.query_params
    storage_filter = Q()
    if storage_service := params["storage_service"]:
        file_id_type = "storage_identifier"
        storage_filter = Q(storage__storage_service=storage_service)

    try:
        if params["relations"]:
            # Return dict of file ids -> list of dataset ids
            file_id_filter = Q(**{f"{file_id_type}__in": ids})
            queryset = File.objects.filter(
                file_id_filter, storage_filter, file_sets__isnull=False
            ).values(key=F(file_id_type))
            queryset = queryset.annotate(
                values=ArrayAgg(
                    "file_sets__dataset_id",
                    filter=Q(file_sets__dataset__deprecated__isnull=True),
                    default=Value([]),
                )
            )
            return Response({str(v["key"]): v["values"] for v in queryset})
        else:
            # Return list of dataset ids
            file_id_filter = Q(**{f"files__{file_id_type}__in": ids})
            file_set_model = File.file_sets.rel.related_model
            queryset = (
                file_set_model.objects.filter(file_id_filter, storage_filter)
                .values_list("dataset__id", flat=True)
                .distinct()
            )
            return Response(queryset)

    except DjangoValidationError as e:
        raise serializers.ValidationError(e.messages)  # avoid 500 from invalid uuid

destroy_list(*args, **kwargs) #

Delete files matching query parameters.

The csc_project parameter is required. If no storage_service is defined, matching files from all storage services are deleted.

By default the files are flagged as removed. When flush is enabled, the files are removed from the database.

Source code in src/apps/files/views/file_view.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
@swagger_auto_schema(
    operation_id="v3_files_delete_list",
    manual_parameters=get_filter_openapi_parameters(FileDeleteListFilterSet),
    responses={200: DeleteListReturnValueSerializer()},
)
def destroy_list(self, *args, **kwargs):
    """Delete files matching query parameters.

    The `csc_project` parameter is required. If no `storage_service` is defined,
    matching files from all storage services are deleted.

    By default the files are flagged as removed.
    When flush is enabled, the files are removed from the database.
    """

    flush = self.query_params["flush"]
    queryset: QuerySet
    if flush:
        queryset = File.all_objects
    else:
        queryset = File.available_objects
    queryset = self.filter_queryset(queryset)

    count = queryset.count()
    queryset.delete()

    return Response(DeleteListReturnValueSerializer(instance={"count": count}).data, 200)