def upload()

in backend/model_garden/views/media_asset.py [0:0]

• 44 lines of code
• 9 McCabe index (conditional complexity)

  def upload(self, request):
    validated_attrs = self._validate_request_params(request=request)
    bucket = validated_attrs['bucket']
    files = validated_attrs['files']
    dataset_format = validated_attrs['dataset_format']

    dataset_serializer = DatasetSerializer(data={
      'path': request.data.get('path'),
      'bucket': bucket.pk,
      'dataset_format': dataset_format,
    })
    dataset_serializer.is_valid(raise_exception=True)
    dataset = dataset_serializer.save()

    files_to_upload = []
    for file in files:
      if file.content_type in ('application/zip', 'application/x-zip-compressed'):
        if not zipfile.is_zipfile(file.file):
          raise ParseError(detail={"message": f"File '{file.name}' is not a zip file"})

        zip_file = zipfile.ZipFile(file.file)
        for zip_filename in zip_file.filelist:
          if zip_filename.filename.endswith('/'):
            continue
          with zip_file.open(zip_filename) as fp:
            files_to_upload.append((zip_filename.filename, io.BytesIO(fp.read())))
      elif 'image' in file.content_type:
        files_to_upload.append((file.name, file.file))
      else:
        logger.warning(f"Got unexpected file content type: {file.content_type}")
        continue

    media_assets_to_upload = []
    for filename, file_obj in files_to_upload:
      try:
        media_asset = MediaAsset.objects.create(dataset=dataset, filename=filename)
      except IntegrityError as e:
        logger.error(f"Failed to create media asset: {e}")
        raise ParseError(
          detail={'message': f"Media asset for dataset='{dataset.path}' and filename='{filename}' already exists"},
        )
      else:
        media_assets_to_upload.append((media_asset, file_obj))

    self._upload_media_assets_to_s3(
      bucket_name=bucket.name,
      media_assets_to_upload=media_assets_to_upload,
    )

    return Response(data={'message': f"{len(media_assets_to_upload)} media asset(s) uploaded"})