Skip to content

Update EntityCsv.java #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wilmer999 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Update EntityCsv.java #1

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 34 additions & 34 deletions openmetadata-service/src/main/java/org/openmetadata/csv/EntityCsv.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,8 @@ public final CsvImportResult importCsv(String csv, boolean dryRun) throws IOExce

// Validate and load each record
while (records.hasNext()) {
CSVRecord record = records.next();
processRecord(resultsPrinter, expectedHeaders, record);
CSVRecord recordCvs = records.next();
processRecord(resultsPrinter, expectedHeaders, recordCvs);
}

// Finally, create the entities parsed from the record
Expand All @@ -108,7 +108,7 @@ public final CsvImportResult importCsv(String csv, boolean dryRun) throws IOExce
}

/** Implement this method to validate each record */
protected abstract T toEntity(CSVPrinter resultsPrinter, CSVRecord record) throws IOException;
protected abstract T toEntity(CSVPrinter resultsPrinter, CSVRecord recordCvs) throws IOException;

public final String exportCsv(List<T> entities) throws IOException {
CsvFile csvFile = new CsvFile().withHeaders(csvHeaders);
Expand All @@ -124,9 +124,9 @@ public final String exportCsv(List<T> entities) throws IOException {
protected abstract List<String> toRecord(T entity);

protected final EntityReference getEntityReference(
CSVPrinter printer, CSVRecord record, int fieldNumber, String entityType) throws IOException {
String fqn = record.get(fieldNumber);
return getEntityReference(printer, record, fieldNumber, entityType, fqn);
CSVPrinter printer, CSVRecord recordCvs, int fieldNumber, String entityType) throws IOException {
String fqn = recordCvs.get(fieldNumber);
return getEntityReference(printer, recordCvs, fieldNumber, entityType, fqn);
}

private EntityInterface getEntity(String entityType, String fqn) {
Expand All @@ -139,29 +139,29 @@ private EntityInterface getEntity(String entityType, String fqn) {
}

protected final EntityReference getEntityReference(
CSVPrinter printer, CSVRecord record, int fieldNumber, String entityType, String fqn) throws IOException {
CSVPrinter printer, CSVRecord recordCvs, int fieldNumber, String entityType, String fqn) throws IOException {
if (nullOrEmpty(fqn)) {
return null;
}
EntityInterface entity = getEntity(entityType, fqn);
if (entity == null) {
importFailure(printer, entityNotFound(fieldNumber, fqn), record);
importFailure(printer, entityNotFound(fieldNumber, fqn), recordCvs);
processRecord = false;
return null;
}
return entity.getEntityReference();
}

protected final List<EntityReference> getEntityReferences(
CSVPrinter printer, CSVRecord record, int fieldNumber, String entityType) throws IOException {
String fqns = record.get(fieldNumber);
CSVPrinter printer, CSVRecord recordCvs, int fieldNumber, String entityType) throws IOException {
String fqns = recordCvs.get(fieldNumber);
if (nullOrEmpty(fqns)) {
return null;
}
List<String> fqnList = listOrEmpty(CsvUtil.fieldToStrings(fqns));
List<EntityReference> refs = new ArrayList<>();
for (String fqn : fqnList) {
EntityReference ref = getEntityReference(printer, record, fieldNumber, entityType, fqn);
EntityReference ref = getEntityReference(printer, recordCvs, fieldNumber, entityType, fqn);
if (!processRecord) {
return null;
}
Expand All @@ -172,9 +172,9 @@ protected final List<EntityReference> getEntityReferences(
return refs.isEmpty() ? null : refs;
}

protected final List<TagLabel> getTagLabels(CSVPrinter printer, CSVRecord record, int fieldNumber)
protected final List<TagLabel> getTagLabels(CSVPrinter printer, CSVRecord recordCvs, int fieldNumber)
throws IOException {
List<EntityReference> refs = getEntityReferences(printer, record, fieldNumber, Entity.TAG);
List<EntityReference> refs = getEntityReferences(printer, recordCvs, fieldNumber, Entity.TAG);
if (!processRecord || nullOrEmpty(refs)) {
return null;
}
Expand Down Expand Up @@ -212,49 +212,49 @@ private Iterator<CSVRecord> parse(String csv) {
return null;
}

private boolean validateHeaders(List<String> expectedHeaders, CSVRecord record) {
importResult.withNumberOfRowsProcessed((int) record.getRecordNumber());
if (expectedHeaders.equals(record.toList())) {
private boolean validateHeaders(List<String> expectedHeaders, CSVRecord recordCvs) {
importResult.withNumberOfRowsProcessed((int) recordCvs.getRecordNumber());
if (expectedHeaders.equals(recordCvs.toList())) {
return true;
}
importResult.withNumberOfRowsFailed(1);
documentFailure(invalidHeader(recordToString(expectedHeaders), recordToString(record)));
documentFailure(invalidHeader(recordToString(expectedHeaders), recordToString(recordCvs)));
return false;
}

private void processRecord(CSVPrinter resultsPrinter, List<String> expectedHeader, CSVRecord record)
private void processRecord(CSVPrinter resultsPrinter, List<String> expectedHeader, CSVRecord recordCvs)
throws IOException {
processRecord = true;
// Every row must have total fields corresponding to the number of headers
if (csvHeaders.size() != record.size()) {
importFailure(resultsPrinter, invalidFieldCount(expectedHeader.size(), record.size()), record);
if (csvHeaders.size() != recordCvs.size()) {
importFailure(resultsPrinter, invalidFieldCount(expectedHeader.size(), recordCvs.size()), recordCvs);
return;
}

// Check if required values are present
List<String> errors = new ArrayList<>();
for (int i = 0; i < csvHeaders.size(); i++) {
String field = record.get(i);
String field = recordCvs.get(i);
boolean fieldRequired = Boolean.TRUE.equals(csvHeaders.get(i).getRequired());
if (fieldRequired && nullOrEmpty(field)) {
errors.add(fieldRequired(i));
}
}

if (!errors.isEmpty()) {
importFailure(resultsPrinter, String.join(FIELD_SEPARATOR, errors), record);
importFailure(resultsPrinter, String.join(FIELD_SEPARATOR, errors), recordCvs);
return;
}

// Finally, convert record into entity for importing
T entity = toEntity(resultsPrinter, record);
T entity = toEntity(resultsPrinter, recordCvs);
if (entity != null) {
// Finally, create entities
createEntity(resultsPrinter, record, entity);
createEntity(resultsPrinter, recordCvs, entity);
}
}

private void createEntity(CSVPrinter resultsPrinter, CSVRecord record, T entity) throws IOException {
private void createEntity(CSVPrinter resultsPrinter, CSVRecord recordCvs, T entity) throws IOException {
entity.setId(UUID.randomUUID());
entity.setUpdatedBy(user);
entity.setUpdatedAt(System.currentTimeMillis());
Expand All @@ -266,7 +266,7 @@ private void createEntity(CSVPrinter resultsPrinter, CSVRecord record, T entity)
PutResponse<EntityInterface> response = repository.createOrUpdate(null, entity);
responseStatus = response.getStatus();
} catch (Exception ex) {
importFailure(resultsPrinter, ex.getMessage(), record);
importFailure(resultsPrinter, ex.getMessage(), recordCvs);
return;
}
} else {
Expand All @@ -280,9 +280,9 @@ private void createEntity(CSVPrinter resultsPrinter, CSVRecord record, T entity)
}

if (Response.Status.CREATED.equals(responseStatus)) {
importSuccess(resultsPrinter, record, ENTITY_CREATED);
importSuccess(resultsPrinter, recordCvs, ENTITY_CREATED);
} else {
importSuccess(resultsPrinter, record, ENTITY_UPDATED);
importSuccess(resultsPrinter, recordCvs, ENTITY_UPDATED);
}
}

Expand Down Expand Up @@ -319,17 +319,17 @@ private void documentFailure(String error) {
}

private void importSuccess(CSVPrinter printer, CSVRecord inputRecord, String successDetails) throws IOException {
List<String> record = listOf(IMPORT_STATUS_SUCCESS, successDetails);
record.addAll(inputRecord.toList());
printer.printRecord(record);
List<String> recordCvs = listOf(IMPORT_STATUS_SUCCESS, successDetails);
recordCvs.addAll(inputRecord.toList());
printer.printRecord(recordCvs);
importResult.withNumberOfRowsProcessed((int) inputRecord.getRecordNumber());
importResult.withNumberOfRowsPassed(importResult.getNumberOfRowsPassed() + 1);
}

protected void importFailure(CSVPrinter printer, String failedReason, CSVRecord inputRecord) throws IOException {
List<String> record = listOf(IMPORT_STATUS_FAILED, failedReason);
record.addAll(inputRecord.toList());
printer.printRecord(record);
List<String> recordCvs = listOf(IMPORT_STATUS_FAILED, failedReason);
recordCvs.addAll(inputRecord.toList());
printer.printRecord(recordCvs);
importResult.withNumberOfRowsProcessed((int) inputRecord.getRecordNumber());
importResult.withNumberOfRowsFailed(importResult.getNumberOfRowsFailed() + 1);
processRecord = false;
Expand Down