@ -97,8 +103,8 @@ export async function importData({
services: AirtableImportContext;
services: AirtableImportContext;
}):Promise<EntityMap>{
}):Promise<EntityMap>{
try{
try{
// @ts-ignore
// returns EntityMap which allows us to stream data
constrecords=awaitreadAllData({
constrecords: EntityMap=awaitreadAllData({
table,
table,
base,
base,
logDetailed,
logDetailed,
@ -109,10 +115,14 @@ export async function importData({
constreadable=records.getStream();
constreadable=records.getStream();
constallRecordsCount=awaitrecords.getCount();
constallRecordsCount=awaitrecords.getCount();
constpromises=[];
constpromises=[];
lettempData=[];
lettempData=[];
letimportedCount=0;
letimportedCount=0;
letactiveProcess=0;
lettempCount=0;
lettempCount=0;
// we keep track of active process to pause and resume the stream as we have async calls within the stream and we don't want to load all data in memory
letactiveProcess=0;
readable.on('data',async(record)=>{
readable.on('data',async(record)=>{
promises.push(
promises.push(
newPromise(async(resolve)=>{
newPromise(async(resolve)=>{
@ -148,6 +158,7 @@ export async function importData({
allRecordsCount,
allRecordsCount,
)}`,
)}`,
);
);
importedCount+=insertArray.length;
importedCount+=insertArray.length;
insertArray=[];
insertArray=[];
@ -162,7 +173,10 @@ export async function importData({
);
);
});
});
readable.on('end',async()=>{
readable.on('end',async()=>{
// ensure all chunks are processed
awaitPromise.all(promises);
awaitPromise.all(promises);
// insert remaining data
if(tempData.length>0){
if(tempData.length>0){
awaitservices.bulkDataService.bulkDataInsert({
awaitservices.bulkDataService.bulkDataInsert({
projectName,
projectName,
@ -182,6 +196,7 @@ export async function importData({
importedCount+=tempData.length;
importedCount+=tempData.length;
tempData=[];
tempData=[];
}
}
resolve(true);
resolve(true);
});
});
});
});
@ -230,7 +245,7 @@ export async function importLTARData({
curCol:{title?: string};
curCol:{title?: string};
refCol:{title?: string};
refCol:{title?: string};
}>=[];
}>=[];
constallData=
constallData: EntityMap=
records||
records||
(awaitreadAllData({
(awaitreadAllData({
table,
table,
@ -288,12 +303,13 @@ export async function importLTARData({
forawait(constassocMetaofassocTableMetas){
forawait(constassocMetaofassocTableMetas){
letassocTableData=[];
letassocTableData=[];
letimportedCount=0;
letimportedCount=0;
lettempCount=0;
// extract insert data from records
// extract link data from records
awaitnewPromise((resolve)=>{
awaitnewPromise((resolve)=>{
constpromises=[];
constpromises=[];
constreadable=allData.getStream();
constreadable=allData.getStream();
lettempCount=0;
readable.on('data',async(record)=>{
readable.on('data',async(record)=>{
promises.push(
promises.push(
newPromise(async(resolve)=>{
newPromise(async(resolve)=>{
@ -318,6 +334,7 @@ export async function importLTARData({
0,
0,
assocTableData.length,
assocTableData.length,
);
);
logBasic(
logBasic(
`:: Importing '${
`:: Importing '${
table.title
table.title
@ -346,7 +363,10 @@ export async function importLTARData({
);
);
});
});
readable.on('end',async()=>{
readable.on('end',async()=>{
// ensure all chunks are processed
awaitPromise.all(promises);
awaitPromise.all(promises);
// insert remaining data
if(assocTableData.length>=0){
if(assocTableData.length>=0){
logBasic(
logBasic(
`:: Importing '${
`:: Importing '${
@ -367,6 +387,7 @@ export async function importLTARData({