-
Notifications
You must be signed in to change notification settings - Fork 429
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[import] Refactored to alllow importing from tarballs
- Loading branch information
Showing
11 changed files
with
248 additions
and
127 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,76 +1,18 @@ | ||
| const debug = require('debug')('sanity:import') | ||
| const flatten = require('lodash/flatten') | ||
| const fromStream = require('./importFromStream') | ||
| const fromFolder = require('./importFromFolder') | ||
| const fromArray = require('./importFromArray') | ||
| const validateOptions = require('./validateOptions') | ||
| const streamToArray = require('./streamToArray') | ||
| const {getAssetRefs, unsetAssetRefs} = require('./assetRefs') | ||
| const assignArrayKeys = require('./assignArrayKeys') | ||
| const assignDocumentId = require('./assignDocumentId') | ||
| const uploadAssets = require('./uploadAssets') | ||
| const documentHasErrors = require('./documentHasErrors') | ||
| const batchDocuments = require('./batchDocuments') | ||
| const importBatches = require('./importBatches') | ||
| const { | ||
| getStrongRefs, | ||
| weakenStrongRefs, | ||
| setTypeOnReferences, | ||
| strengthenReferences | ||
| } = require('./references') | ||
|
|
||
| async function importDocuments(input, opts) { | ||
| const options = validateOptions(input, opts) | ||
|
|
||
| options.onProgress({step: 'Reading/validating data file'}) | ||
| const isStream = typeof input.pipe === 'function' | ||
| let documents = input | ||
| if (isStream) { | ||
| debug('Streaming input source to array of documents') | ||
| documents = await streamToArray(input) | ||
| } else { | ||
| documents.some(documentHasErrors.validate) | ||
| } | ||
|
|
||
| // Assign document IDs for document that do not have one. This is necessary | ||
| // for us to strengthen references and import assets properly. | ||
| const ided = documents.map(doc => assignDocumentId(doc)) | ||
|
|
||
| // User might not have applied `_key` on array elements which are objects; | ||
| // if this is the case, generate random keys to help realtime engine | ||
| const keyed = ided.map(doc => assignArrayKeys(doc)) | ||
|
|
||
| // Sanity prefers to have a `_type` on every object. Make sure references | ||
| // has `_type` set to `reference`. | ||
| const docs = keyed.map(doc => setTypeOnReferences(doc)) | ||
|
|
||
| // Find references that will need strengthening when import is done | ||
| const strongRefs = docs.map(getStrongRefs).filter(Boolean) | ||
|
|
||
| // Extract asset references from the documents | ||
| const assetRefs = flatten(docs.map(getAssetRefs).filter(ref => ref.length)) | ||
|
|
||
| // Remove asset references from the documents | ||
| const assetless = docs.map(unsetAssetRefs) | ||
|
|
||
| // Make strong references weak so they can be imported in any order | ||
| const weakened = assetless.map(weakenStrongRefs) | ||
|
|
||
| // Create batches of documents to import. Try to keep batches below a certain | ||
| // byte-size (since document may vary greatly in size depending on type etc) | ||
| const batches = batchDocuments(weakened) | ||
|
|
||
| // Trigger actual import process | ||
| debug('Starting import of documents') | ||
| const docsImported = await importBatches(batches, options) | ||
|
|
||
| // Documents are imported, now proceed with post-import operations | ||
| debug('Uploading assets') | ||
| await uploadAssets(assetRefs, options) | ||
| const importers = { | ||
| fromStream, | ||
| fromFolder, | ||
| fromArray | ||
| } | ||
|
|
||
| // Strengthen references | ||
| debug('Strengthening references') | ||
| await strengthenReferences(strongRefs, options) | ||
| module.exports = (input, opts) => { | ||
| const options = validateOptions(input, opts) | ||
|
|
||
| // Return number of documents imported | ||
| return docsImported | ||
| return Array.isArray(input) | ||
| ? fromArray(input, options, importers) | ||
| : fromStream(input, options, importers) | ||
| } | ||
|
|
||
| module.exports = importDocuments |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| const debug = require('debug')('sanity:import:array') | ||
| const flatten = require('lodash/flatten') | ||
| const {getAssetRefs, unsetAssetRefs, absolutifyPaths} = require('./assetRefs') | ||
| const assignArrayKeys = require('./assignArrayKeys') | ||
| const assignDocumentId = require('./assignDocumentId') | ||
| const uploadAssets = require('./uploadAssets') | ||
| const documentHasErrors = require('./documentHasErrors') | ||
| const batchDocuments = require('./batchDocuments') | ||
| const importBatches = require('./importBatches') | ||
| const { | ||
| getStrongRefs, | ||
| weakenStrongRefs, | ||
| setTypeOnReferences, | ||
| strengthenReferences | ||
| } = require('./references') | ||
|
|
||
| async function importDocuments(documents, options, importers) { | ||
| options.onProgress({step: 'Reading/validating data file'}) | ||
| documents.some(documentHasErrors.validate) | ||
|
|
||
| // Replace relative asset paths if one is defined | ||
| // (file://./images/foo-bar.png -> file:///abs/olute/images/foo-bar.png) | ||
| const absPathed = documents.map(doc => absolutifyPaths(doc, options.assetsBase)) | ||
|
|
||
| // Assign document IDs for document that do not have one. This is necessary | ||
| // for us to strengthen references and import assets properly. | ||
| const ided = absPathed.map(doc => assignDocumentId(doc)) | ||
|
|
||
| // User might not have applied `_key` on array elements which are objects; | ||
| // if this is the case, generate random keys to help realtime engine | ||
| const keyed = ided.map(doc => assignArrayKeys(doc)) | ||
|
|
||
| // Sanity prefers to have a `_type` on every object. Make sure references | ||
| // has `_type` set to `reference`. | ||
| const docs = keyed.map(doc => setTypeOnReferences(doc)) | ||
|
|
||
| // Find references that will need strengthening when import is done | ||
| const strongRefs = docs.map(getStrongRefs).filter(Boolean) | ||
|
|
||
| // Extract asset references from the documents | ||
| const assetRefs = flatten(docs.map(getAssetRefs).filter(ref => ref.length)) | ||
|
|
||
| // Remove asset references from the documents | ||
| const assetless = docs.map(unsetAssetRefs) | ||
|
|
||
| // Make strong references weak so they can be imported in any order | ||
| const weakened = assetless.map(weakenStrongRefs) | ||
|
|
||
| // Create batches of documents to import. Try to keep batches below a certain | ||
| // byte-size (since document may vary greatly in size depending on type etc) | ||
| const batches = batchDocuments(weakened) | ||
|
|
||
| // Trigger actual import process | ||
| debug('Starting import of documents') | ||
| const docsImported = await importBatches(batches, options) | ||
|
|
||
| // Documents are imported, now proceed with post-import operations | ||
| debug('Uploading assets') | ||
| await uploadAssets(assetRefs, options) | ||
|
|
||
| // Strengthen references | ||
| debug('Strengthening references') | ||
| await strengthenReferences(strongRefs, options) | ||
|
|
||
| // Return number of documents imported | ||
| return docsImported | ||
| } | ||
|
|
||
| module.exports = importDocuments |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| const fse = require('fs-extra') | ||
| const globby = require('globby') | ||
| const debug = require('debug')('sanity:import:folder') | ||
|
|
||
| module.exports = async function importFromFolder(fromDir, options, importers) { | ||
| debug('Importing from folder %s', fromDir) | ||
| const dataFiles = await globby('*.ndjson', {cwd: fromDir, absolute: true}) | ||
| if (dataFiles.length === 0) { | ||
| throw new Error(`No .ndjson file found in ${fromDir}`) | ||
| } | ||
|
|
||
| if (dataFiles.length > 1) { | ||
| throw new Error(`More than one .ndjson file found in ${fromDir} - only one is supported`) | ||
| } | ||
|
|
||
| const dataFile = dataFiles[0] | ||
| debug('Importing from file %s', dataFile) | ||
|
|
||
| const stream = fse.createReadStream(dataFile) | ||
| const images = await globby('images/*', {cwd: fromDir, absolute: true}) | ||
| const files = await globby('files/*', {cwd: fromDir, absolute: true}) | ||
| const unreferencedAssets = [] | ||
| .concat(images.map(path => `image#${path}`)) | ||
| .concat(files.map(path => `file#${path}`)) | ||
|
|
||
| debug('Queueing %d assets', unreferencedAssets.length) | ||
|
|
||
| const streamOptions = {...options, unreferencedAssets, assetsBase: fromDir} | ||
| const result = await importers.fromStream(stream, streamOptions, importers) | ||
|
|
||
| if (options.deleteOnComplete) { | ||
| await fse.remove(fromDir) | ||
| } | ||
|
|
||
| return result | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| const os = require('os') | ||
| const path = require('path') | ||
| const miss = require('mississippi') | ||
| const gunzipMaybe = require('gunzip-maybe') | ||
| const peek = require('peek-stream') | ||
| const isTar = require('is-tar') | ||
| const tar = require('tar-fs') | ||
| const globby = require('globby') | ||
| const debug = require('debug')('sanity:import:stream') | ||
| const getJsonStreamer = require('./util/getJsonStreamer') | ||
|
|
||
| module.exports = (stream, options, importers) => | ||
| new Promise((resolve, reject) => { | ||
| const outputPath = path.join(os.tmpdir(), 'sanity-import') | ||
| debug('Importing from stream') | ||
|
|
||
| let isTarStream = false | ||
|
|
||
| miss.pipe(stream, gunzipMaybe(), untarMaybe(), err => { | ||
| if (err) { | ||
| reject(err) | ||
| return | ||
| } | ||
|
|
||
| if (!isTarStream) { | ||
| return // Will be resolved by concatenation | ||
| } | ||
|
|
||
| findAndImport() | ||
| }) | ||
|
|
||
| function untarMaybe() { | ||
| return peek({newline: false, maxBuffer: 300}, (data, swap) => { | ||
| if (isTar(data)) { | ||
| debug('Stream is a tarball, extracting to %s', outputPath) | ||
| isTarStream = true | ||
| return swap(null, tar.extract(outputPath)) | ||
| } | ||
|
|
||
| debug('Stream is an ndjson file, streaming JSON') | ||
| return swap(null, miss.pipeline(getJsonStreamer(), miss.concat(resolveNdjsonStream))) | ||
| }) | ||
| } | ||
|
|
||
| function resolveNdjsonStream(documents) { | ||
| debug('Finished reading ndjson stream') | ||
| resolve(importers.fromArray(documents, options)) | ||
| } | ||
|
|
||
| async function findAndImport() { | ||
| debug('Tarball extracted, looking for ndjson') | ||
|
|
||
| const files = await globby('**/*.ndjson', {cwd: outputPath, deep: 2, absolute: true}) | ||
| if (!files.length) { | ||
| reject(new Error('ndjson-file not found in tarball')) | ||
| return | ||
| } | ||
|
|
||
| const importBaseDir = path.dirname(files[0]) | ||
| resolve(importers.fromFolder(importBaseDir, {...options, deleteOnComplete: true}, importers)) | ||
| } | ||
| }) |
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.