Skip to content

Commit 9d481ea

Browse files
authorjappsauthorjapps
authored
Merge pull request #686 from javiertuya/680-db-sql-executor
ISSUE-680 # Add the DB SQL Executor to import data from CSV and execute SQL statements
2 parents c86bb83 + ce8f2aa commit 9d481ea

25 files changed

+1363
-6
lines changed

‎.github/workflows/main.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,5 +28,8 @@ jobs:
2828
- name: Running Kafka
2929
run: docker-compose -f docker/compose/kafka-schema-registry.yml up -d && sleep 10
3030

31+
- name: Running PostgreSQL (to test DB SQL Executor)
32+
run: docker-compose -f docker/compose/pg_compose.yml up -d
33+
3134
- name: Building and testing the changes
3235
run: mvn clean test

‎core/pom.xml‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -175,11 +175,20 @@
175175
<artifactId>micro-simulator</artifactId>
176176
<scope>test</scope>
177177
</dependency>
178+
<dependency>
179+
<groupId>commons-dbutils</groupId>
180+
<artifactId>commons-dbutils</artifactId>
181+
</dependency>
178182
<dependency>
179183
<groupId>com.h2database</groupId>
180184
<artifactId>h2</artifactId>
181185
<scope>test</scope>
182186
</dependency>
187+
<dependency>
188+
<groupId>org.postgresql</groupId>
189+
<artifactId>postgresql</artifactId>
190+
<!--<scope>test</scope>--> <!-- Make it available to dependant projects. Hence commented -->
191+
</dependency>
183192
<dependency>
184193
<groupId>com.aventstack</groupId>
185194
<artifactId>extentreports</artifactId>

‎core/src/main/java/org/jsmart/zerocode/core/db/DbCsvLoader.java‎

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
package org.jsmart.zerocode.core.db;
2+
3+
import java.sql.Connection;
4+
import java.sql.SQLException;
5+
import java.util.ArrayList;
6+
import java.util.Arrays;
7+
import java.util.List;
8+
import java.util.stream.Collectors;
9+
import java.util.stream.IntStream;
10+
11+
import org.apache.commons.dbutils.QueryRunner;
12+
import org.apache.commons.lang3.StringUtils;
13+
import org.slf4j.Logger;
14+
import org.slf4j.LoggerFactory;
15+
16+
import com.univocity.parsers.csv.CsvParser;
17+
18+
/**
19+
* Data loading in the database from a CSV external source
20+
*/
21+
class DbCsvLoader {
22+
private static final Logger LOGGER = LoggerFactory.getLogger(DbCsvLoader.class);
23+
private Connection conn;
24+
private CsvParser csvParser;
25+
26+
public DbCsvLoader(Connection conn, CsvParser csvParser) {
27+
this.conn = conn;
28+
this.csvParser = csvParser;
29+
}
30+
31+
/**
32+
* Loads rows in CSV format (csvLines) into a table in the database
33+
* and returns the total number of rows.
34+
*/
35+
public int loadCsv(String table, List<String> csvLines, boolean withHeaders, String nullString) throws SQLException {
36+
if (csvLines == null || csvLines.isEmpty())
37+
return 0;
38+
39+
List<String[]> lines = parseLines(table, csvLines);
40+
41+
String[] headers = buildHeaders(lines.get(0), withHeaders);
42+
List<Object[]> paramset = buildParameters(table, headers, lines, withHeaders, nullString);
43+
if (paramset.isEmpty()) // can have headers, but no rows
44+
return 0;
45+
46+
String sql = buildSql(table, headers, paramset.get(0).length);
47+
LOGGER.info("Loading CSV using this sql: {}", sql);
48+
49+
QueryRunner runner = new QueryRunner();
50+
int insertCount = 0;
51+
for (int i = 0 ; i < paramset.size(); i++) {
52+
insertRow(runner, i, sql, paramset.get(i));
53+
insertCount++;
54+
}
55+
LOGGER.info("Total of rows inserted: {}", insertCount);
56+
return insertCount;
57+
}
58+
59+
private List<String[]> parseLines(String table, List<String> lines) {
60+
int numCol = 0; // will check that every row has same columns than the first
61+
List<String[]> parsedLines = new ArrayList<>();
62+
for (int i = 0; i<lines.size(); i++) {
63+
String[] parsedLine = csvParser.parseLine(lines.get(i));
64+
parsedLines.add(parsedLine);
65+
if (i == 0) {
66+
numCol=parsedLine.length;
67+
} else if (numCol != parsedLine.length) {
68+
String message = String.format("Error parsing CSV content to load into table %s: "
69+
+ "Row %d has %d columns and should have %d", table, i + 1, parsedLine.length, numCol);
70+
LOGGER.error(message);
71+
throw new RuntimeException(message);
72+
}
73+
}
74+
return parsedLines;
75+
}
76+
77+
private String[] buildHeaders(String[] line, boolean withHeaders) {
78+
return withHeaders ? line : new String[] {};
79+
}
80+
81+
private List<Object[]> buildParameters(String table, String[] headers, List<String[]> lines, boolean withHeaders, String nullString) {
82+
DbValueConverter converter = new DbValueConverter(conn, table);
83+
List<Object[]> paramset = new ArrayList<>();
84+
for (int i = withHeaders ? 1 : 0; i < lines.size(); i++) {
85+
String[] parsedLine = lines.get(i);
86+
parsedLine = processNulls(parsedLine, nullString);
87+
Object[] params;
88+
try {
89+
params = converter.convertColumnValues(headers, parsedLine);
90+
LOGGER.info(" row [{}] params: {}", i + 1, Arrays.asList(params).toString());
91+
} catch (Exception e) { // Not only SQLException as converter also does parsing
92+
String message = String.format("Error matching data type of parameters and table columns at CSV row %d", i + 1);
93+
LOGGER.error(message);
94+
LOGGER.error("Exception message: {}", e.getMessage());
95+
throw new RuntimeException(message, e);
96+
}
97+
paramset.add(params);
98+
}
99+
return paramset;
100+
}
101+
102+
private String[] processNulls(String[] line, String nullString) {
103+
for (int i = 0; i < line.length; i++) {
104+
if (StringUtils.isBlank(nullString) && StringUtils.isBlank(line[i])) {
105+
line[i] = null;
106+
} else if (!StringUtils.isBlank(nullString)) {
107+
if (StringUtils.isBlank(line[i])) // null must be empty string
108+
line[i] = "";
109+
else if (nullString.trim().equalsIgnoreCase(line[i].trim()))
110+
line[i] = null;
111+
}
112+
}
113+
return line;
114+
}
115+
116+
private String buildSql(String table, String[] headers, int columnCount) {
117+
String placeholders = IntStream.range(0, columnCount)
118+
.mapToObj(i -> "?").collect(Collectors.joining(","));
119+
return "INSERT INTO " + table
120+
+ (headers.length > 0 ? " (" + String.join(",", headers) + ")" : "")
121+
+ " VALUES (" + placeholders + ");";
122+
}
123+
124+
private void insertRow(QueryRunner runner, int rowId, String sql, Object[] params) {
125+
try {
126+
runner.update(conn, sql, params);
127+
} catch (SQLException e) {
128+
String message = String.format("Error inserting data at CSV row %d", rowId + 1);
129+
LOGGER.error(message);
130+
LOGGER.error("Exception message: {}", e.getMessage());
131+
throw new RuntimeException(message, e);
132+
}
133+
}
134+
135+
}

‎core/src/main/java/org/jsmart/zerocode/core/db/DbCsvRequest.java‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
package org.jsmart.zerocode.core.db;
2+
3+
import com.fasterxml.jackson.annotation.JsonProperty;
4+
import com.fasterxml.jackson.core.type.TypeReference;
5+
import com.fasterxml.jackson.databind.JsonNode;
6+
import com.fasterxml.jackson.databind.ObjectMapper;
7+
import com.fasterxml.jackson.databind.ObjectReader;
8+
import org.apache.commons.lang3.StringUtils;
9+
10+
import java.io.IOException;
11+
import java.nio.file.Files;
12+
import java.nio.file.Path;
13+
import java.nio.file.Paths;
14+
import java.util.Collections;
15+
import java.util.List;
16+
import java.util.Optional;
17+
import java.util.stream.Collectors;
18+
19+
public class DbCsvRequest {
20+
private final String tableName;
21+
private final List<String> csvSource;
22+
private final Boolean withHeaders;
23+
private final String nullString;
24+
25+
public DbCsvRequest(
26+
@JsonProperty(value="tableName", required=true) String tableName,
27+
@JsonProperty("csvSource") JsonNode csvSourceJsonNode,
28+
@JsonProperty("withHeaders") Boolean withHeaders,
29+
@JsonProperty("nullString") String nullString) {
30+
this.tableName = tableName;
31+
this.withHeaders = Optional.ofNullable(withHeaders).orElse(false);
32+
this.nullString = Optional.ofNullable(nullString).orElse("");
33+
this.csvSource = Optional.ofNullable(csvSourceJsonNode).map(this::getCsvSourceFrom).orElse(Collections.emptyList());
34+
}
35+
36+
public String getTableName() {
37+
return tableName;
38+
}
39+
40+
public List<String> getCsvSource() {
41+
return csvSource;
42+
}
43+
44+
public boolean getWithHeaders() {
45+
return withHeaders;
46+
}
47+
48+
public String getNullString() {
49+
return nullString;
50+
}
51+
52+
// Code below is duplicated from org.jsmart.zerocode.core.domain.Parametrized.java and not included in tests.
53+
// TODO Consider some refactoring later and review error message when file not found
54+
55+
private List<String> getCsvSourceFrom(JsonNode csvSourceJsonNode) {
56+
try {
57+
if (csvSourceJsonNode.isArray()) {
58+
return readCsvSourceFromJson(csvSourceJsonNode);
59+
60+
} else {
61+
return readCsvSourceFromExternalCsvFile(csvSourceJsonNode);
62+
}
63+
} catch (IOException e) {
64+
throw new RuntimeException("Error deserializing csvSource", e);
65+
}
66+
}
67+
68+
private List<String> readCsvSourceFromJson(JsonNode csvSourceJsonNode) throws IOException {
69+
ObjectMapper mapper = new ObjectMapper();
70+
ObjectReader reader = mapper.readerFor(new TypeReference<List<String>>() {
71+
});
72+
return reader.readValue(csvSourceJsonNode);
73+
}
74+
75+
private List<String> readCsvSourceFromExternalCsvFile(JsonNode csvSourceJsonNode) throws IOException {
76+
String csvSourceFilePath = csvSourceJsonNode.textValue();
77+
if (StringUtils.isNotBlank(csvSourceFilePath)) {
78+
Path path = Paths.get("./src/test/resources/",csvSourceFilePath);
79+
List<String> csvSourceFileLines = Files.lines(path)
80+
.filter(StringUtils::isNotBlank)
81+
.collect(Collectors.toList());
82+
//if (this.ignoreHeader) {
83+
// return csvSourceFileLines.stream()
84+
// .skip(1)
85+
// .collect(Collectors.toList());
86+
//}
87+
return csvSourceFileLines;
88+
}
89+
return Collections.emptyList();
90+
}
91+
92+
@Override
93+
public String toString() {
94+
return "Parameterized{" +
95+
"tableName=" + tableName +
96+
", csvSource=" + csvSource +
97+
", withHeaders=" + withHeaders +
98+
", nullString=" + nullString +
99+
'}';
100+
}
101+
}

‎core/src/main/java/org/jsmart/zerocode/core/db/DbSqlExecutor.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package org.jsmart.zerocode.core.db;
2+
3+
import com.fasterxml.jackson.databind.ObjectMapper;
4+
import com.google.inject.Inject;
5+
import com.google.inject.name.Named;
6+
import com.univocity.parsers.csv.CsvParser;
7+
8+
import org.apache.commons.dbutils.DbUtils;
9+
import org.slf4j.Logger;
10+
import org.slf4j.LoggerFactory;
11+
import java.sql.Connection;
12+
import java.sql.DriverManager;
13+
import java.sql.SQLException;
14+
import java.util.HashMap;
15+
import java.util.List;
16+
import java.util.Map;
17+
18+
/**
19+
* Interaction with a database using SQL to read/write
20+
* Requires the appropriated connection data in the target environment
21+
* properties, see src/test/resources/db_test.properties
22+
*/
23+
public class DbSqlExecutor {
24+
private static final Logger LOGGER = LoggerFactory.getLogger(DbSqlExecutor.class);
25+
public static final String SQL_RESULTS_KEY = "rows";
26+
public static final String CSV_RESULTS_KEY = "size";
27+
28+
// Optional to log the explanatory error message if the env variables are no defined
29+
@Inject(optional = true)
30+
@Named("db.driver.url") private String url;
31+
32+
@Inject(optional = true)
33+
@Named("db.driver.user") private String user;
34+
35+
@Inject(optional = true)
36+
@Named("db.driver.password") private String password;
37+
38+
@Inject
39+
private CsvParser csvParser;
40+
41+
/**
42+
* The LOADCSV operation inserts the content of a CSV file into a table,
43+
* and returns the number of records inserted under the key "size"
44+
*/
45+
public Map<String, Object> LOADCSV(DbCsvRequest request) { // uppercase for consistency with http api operations
46+
return loadcsv(request);
47+
}
48+
49+
public Map<String, Object> loadcsv(DbCsvRequest request) {
50+
Connection conn = createAndGetConnection();
51+
try {
52+
LOGGER.info("Load CSV, request -> {} ", request);
53+
DbCsvLoader runner = new DbCsvLoader(conn, csvParser);
54+
long result = runner.loadCsv(request.getTableName(), request.getCsvSource(),
55+
request.getWithHeaders(), request.getNullString());
56+
Map<String, Object> response = new HashMap<>();
57+
response.put(CSV_RESULTS_KEY, result);
58+
return response;
59+
} catch (Exception e) {
60+
String message = "Failed to load CSV";
61+
LOGGER.error(message, e);
62+
throw new RuntimeException(message, e);
63+
} finally {
64+
closeConnection(conn);
65+
}
66+
}
67+
68+
/**
69+
* The EXECUTE operation returns the records retrieved by the SQL specified in the request
70+
* under the key "rows" (select), or an empty object (insert, update)
71+
*/
72+
public Map<String, Object> EXECUTE(DbSqlRequest request) {
73+
return execute(request);
74+
}
75+
76+
public Map<String, Object> execute(DbSqlRequest request) {
77+
Connection conn = createAndGetConnection();
78+
try {
79+
LOGGER.info("Execute SQL, request -> {} ", request);
80+
DbSqlRunner runner = new DbSqlRunner(conn);
81+
List<Map<String, Object>> results = runner.execute(request.getSql(), request.getSqlParams());
82+
Map<String, Object> response = new HashMap<>();
83+
if (results == null) { // will return empty node, use "verify":{}
84+
response.put(SQL_RESULTS_KEY, new ObjectMapper().createObjectNode());
85+
} else {
86+
response.put(SQL_RESULTS_KEY, results);
87+
}
88+
return response;
89+
} catch (SQLException e) {
90+
String message = "Failed to execute SQL";
91+
LOGGER.error(message, e);
92+
throw new RuntimeException(message, e);
93+
} finally {
94+
closeConnection(conn);
95+
}
96+
}
97+
98+
/**
99+
* Returns a new JDBC connection using DriverManager.
100+
* Override this method in case you get the connections using another approach
101+
* (e.g. DataSource)
102+
*/
103+
protected Connection createAndGetConnection() {
104+
LOGGER.info("Create and get connection, url: {}, user: {}", url, user);
105+
try {
106+
return DriverManager.getConnection(url, user, password);
107+
} catch (SQLException e) {
108+
String message = "Failed to create connection, Please check the target environment properties "
109+
+ "to connect the database (db.driver.url, db.driver.user and db.driver.password)";
110+
LOGGER.error(message, e);
111+
throw new RuntimeException(message, e);
112+
}
113+
}
114+
115+
protected void closeConnection(Connection conn) {
116+
DbUtils.closeQuietly(conn);
117+
}
118+
119+
}

0 commit comments

Comments
 (0)