Class ImportStorageListFilesAPI

label_studio/io_storages/api.py:192–230

@extend_schema(exclude=True)
class ImportStorageListFilesAPI(generics.CreateAPIView):

    permission_required = all_permissions.storages_change
    parser_classes = (JSONParser, FormParser, MultiPartParser)
    serializer_class = None  # Default serializer

    def __init__(self, serializer_class=None, *args, **kwargs):
        self.serializer_class = serializer_class
        super().__init__(*args, **kwargs)

    @extend_schema(exclude=True)
    def create(self, request, *args, **kwargs):
        from .functions import validate_storage_instance

        instance = validate_storage_instance(request, self.serializer_class)
        limit = int(request.data.get('limit', settings.DEFAULT_STORAGE_LIST_LIMIT))

        try:
            files = []
            start_time = time.time()
            timeout_seconds = 30

            for object in instance.iter_objects():
                files.append(instance.get_unified_metadata(object))

                # Check if we've reached the file limit
                if len(files) >= limit:
                    files.append({'key': None, 'last_modified': None, 'size': None})
                    break

                # Check if we've exceeded the timeout
                if time.time() - start_time > timeout_seconds:
                    files.append({'key': '... storage scan timeout reached ...', 'last_modified': None, 'size': None})
                    break

            return Response({'files': files})
        except Exception as exc:
            logger.exception('Error listing storage files: %s', exc)
            raise ValidationError('Failed to list storage files')


@extend_schema(exclude=True)