[export] Add concurrency flag to export command

packages/@sanity/core/src/commands/dataset/exportDatasetCommand.js

@@ -10,12 +10,13 @@ const noop = () => null
 
 const helpText = `
 Options
-  --raw         Extract only documents, without rewriting asset references
-  --no-assets   Export only non-asset documents and remove references to image assets
-  --no-drafts   Export only published versions of documents
-  --no-compress Skips compressing tarball entries (still generates a gzip file)
-  --types       Defines which document types to export
-  --overwrite   Overwrite any file with the same name
+  --raw                Extract only documents, without rewriting asset references
+  --no-assets          Export only non-asset documents and remove references to image assets
+  --no-drafts          Export only published versions of documents
+  --no-compress        Skips compressing tarball entries (still generates a gzip file)
+  --types              Defines which document types to export
+  --overwrite          Overwrite any file with the same name
+  --concurrency <field> Concurrent downloads of assets
 
 Examples
   sanity dataset export moviedb localPath.tar.gz

packages/@sanity/export/README.md

@@ -38,7 +38,10 @@ exportDataset({
 
   // Export only given document types (`_type`)
   // Optional, default: all types
-  types: ['products', 'shops']
+  types: ['products', 'shops'],
+
+  // Run 12 concurrent asset downloads
+  concurrency: 12
 })
 ```
 

packages/@sanity/export/src/AssetHandler.js

@@ -12,6 +12,7 @@ const debug = require('./debug')
 const EXCLUDE_PROPS = ['_id', '_type', 'assetId', 'extension', 'mimeType', 'path', 'url']
 const ACTION_REMOVE = 'remove'
 const ACTION_REWRITE = 'rewrite'
+const ASSET_DOWNLOAD_CONCURRENCY = 8
 
 class AssetHandler {
   constructor(options) {
@@ -23,7 +24,7 @@ class AssetHandler {
     this.assetMap = {}
     this.filesWritten = 0
     this.queueSize = 0
-    this.queue = options.queue || new PQueue({concurrency: 3})
+    this.queue = options.queue || new PQueue({concurrency: options.concurrency || ASSET_DOWNLOAD_CONCURRENCY})
     this.rejectedError = null
     this.reject = err => {
       this.rejectedError = err

packages/@sanity/export/src/export.js

@@ -41,7 +41,8 @@ function exportDataset(opts) {
   const assetHandler = new AssetHandler({
     client: options.client,
     tmpDir,
-    prefix
+    prefix,
+    concurrency: options.concurrency
   })
 
   debug('Outputting assets (temporarily) to %s', tmpDir)

packages/@sanity/export/src/validateOptions.js

@@ -46,6 +46,10 @@ function validateOptions(opts) {
     throw new Error('outputPath must be specified (- for stdout)')
   }
 
+  if (options.concurrency && options.concurrency > 24) {
+    throw new Error('`concurrency` must be <= 24')
+  }
+
   return options
 }
 

packages/@sanity/import-cli/README.md

@@ -20,6 +20,7 @@ npm install -g @sanity/import-cli
     -p, --project <projectId> Project ID to import to
     -d, --dataset <dataset> Dataset to import to
     -t, --token <token> Token to authenticate with
+    --concurrency <concurrency> Number of parallel asset imports
     --replace Replace documents with the same IDs
     --missing Skip documents that already exist
     --help Show this help

packages/@sanity/import-cli/src/sanity-import.js

@@ -22,6 +22,7 @@ const cli = meow(
     -p, --project <projectId> Project ID to import to
     -d, --dataset <dataset> Dataset to import to
     -t, --token <token> Token to authenticate with
+    --concurrency <concurrency> Number of parallel asset imports
     --replace Replace documents with the same IDs
     --missing Skip documents that already exist
     --allow-failing-assets Skip assets that cannot be fetched/uploaded
@@ -42,7 +43,8 @@ const cli = meow(
     alias: {
       p: 'project',
       d: 'dataset',
-      t: 'token'
+      t: 'token',
+      c: 'concurrency'
     }
   }
 )
@@ -51,6 +53,7 @@ const {flags, input, showHelp} = cli
 const {dataset, allowFailingAssets} = flags
 const token = flags.token || process.env.SANITY_IMPORT_TOKEN
 const projectId = flags.project
+const concurrency = flags.concurrency
 const source = input[0]
 
 if (!projectId) {
@@ -96,7 +99,9 @@ const client = sanityClient({
 })
 
 getStream()
-  .then(stream => sanityImport(stream, {client, operation, onProgress, allowFailingAssets}))
+  .then(stream =>
+    sanityImport(stream, {client, operation, onProgress, allowFailingAssets, concurrency})
+  )
   .then(({numDocs, warnings}) => {
     const timeSpent = prettyMs(Date.now() - stepStart, {secDecimalDigits: 2})
     currentProgress.text = `[100%] ${currentStep} (${timeSpent})`

packages/@sanity/import/src/uploadAssets.js

@@ -45,7 +45,7 @@ async function uploadAssets(assets, options) {
     : ensureAssetExists
 
   // Loop over all unique URLs and ensure they exist, and if not, upload them
-  const mapOptions = {concurrency: ASSET_UPLOAD_CONCURRENCY}
+  const mapOptions = {concurrency: options.concurrency || ASSET_UPLOAD_CONCURRENCY}
   const assetIds = await pMap(assetRefMap.keys(), ensureMethod, mapOptions)
 
   // Extract a list of all failures so we may report them and possibly retry them later

packages/@sanity/import/src/validateOptions.js

@@ -39,6 +39,10 @@ function validateOptions(input, opts) {
     throw new Error(`Operation "${options.operation}" is not supported`)
   }
 
+  if (options.concurrency && options.concurrency > 24) {
+    throw new Error('`concurrency` must be <= 24')
+  }
+
   return options
 }