Postgres config 3605

airbyte-config/db-persistence/build.gradle

@@ -0,0 +1,11 @@
+dependencies {
+    implementation group: 'commons-io', name: 'commons-io', version: '2.7'
+
+    implementation project(':airbyte-db')
+    implementation project(':airbyte-commons')
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-config:persistence')
+    implementation project(":airbyte-json-validation")
+
+    testImplementation "org.testcontainers:postgresql:1.15.1"
+}

airbyte-config/db-persistence/src/main/java/io.airbyte.config.dbPersistence/DatabaseConfigPersistence.java

@@ -0,0 +1,118 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.config.dbPersistence;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.ConfigSchema;
+import io.airbyte.config.persistence.ConfigNotFoundException;
+import io.airbyte.config.persistence.ConfigPersistence;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import io.airbyte.validation.json.JsonSchemaValidator;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class DatabaseConfigPersistence implements ConfigPersistence {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(DatabaseConfigPersistence.class);
+
+  protected final JdbcDatabase database;
+  private final JsonSchemaValidator jsonSchemaValidator;
+
+  protected DatabaseConfigPersistence(JdbcDatabase db, JsonSchemaValidator validator) {
+    database = db;
+    jsonSchemaValidator = validator;
+  }
+
+  // Create a table named CONFIG with three columns: CONFIG_ID (PK UUID/string), CONFIG_TYPE (string),
+  // CONFIG_DATA (JSON/string)
+  public abstract void Setup() throws SQLException;
+
+  @Override
+  public <T> T getConfig(ConfigSchema configType, UUID configId, Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    try {
+      Optional<String> data = database.querySingle("SELECT CONFIG_DATA FROM CONFIG WHERE CONFIG_TYPE = ? AND CONFIG_ID = ?",
+          r -> r.getString("CONFIG_DATA"), configType.toString(), configId);
+      if (!data.isPresent()) {
+        throw new ConfigNotFoundException(configType, configId);
+      }
+
+      final T config = Jsons.deserialize(data.get(), clazz);
+      validateJson(config, configType);
+      return config;
+    } catch (SQLException e) {
+      throw new IOException(String.format("Failed to get config type %s item %s.  Reason: %s", configType, configId, e.getMessage()), e);
+    }
+  }
+
+  @Override
+  public <T> List<T> listConfigs(ConfigSchema configType, Class<T> clazz) throws JsonValidationException, IOException {
+    try {
+      List<T> results = database.query(c -> {
+        var stmt = c.prepareStatement("SELECT CONFIG_DATA FROM CONFIG WHERE CONFIG_TYPE = ?");
+        stmt.setString(1, configType.toString());
+        return stmt;
+      }, r -> r.getString("CONFIG_DATA"))
+          .map(s -> (T) Jsons.deserialize(s, clazz))
+          .collect(Collectors.toList());
+      return results;
+    } catch (SQLException e) {
+      throw new IOException(String.format("Failed to get config type %s listing.  Reason: %s", configType, e.getMessage()), e);
+    }
+  }
+
+  @Override
+  public <T> void writeConfig(ConfigSchema configType, UUID configId, T config) throws JsonValidationException, IOException {
+    // validate config with schema
+    validateJson(Jsons.jsonNode(config), configType);
+    final String data = Jsons.serialize(config);
+    try {
+      database.execute(c -> writeConfigQuery(c, configType, configId, data).execute());
+    } catch (SQLException e) {
+      throw new IOException(String.format("Failed to write config type %s item %s.  Reason: %s", configType, configId, e.getMessage()), e);
+    }
+  }
+
+  // Made abstract because what we want for this is an upsert operation, which different databases
+  // handle with different syntax
+  // Overrides need to return a prepared statement with all 3 data elements added
+  protected abstract PreparedStatement writeConfigQuery(Connection conn, ConfigSchema configType, UUID configId, String data) throws SQLException;
+
+  private <T> void validateJson(T config, ConfigSchema configType) throws JsonValidationException {
+    JsonNode schema = JsonSchemaValidator.getSchema(configType.getFile());
+    jsonSchemaValidator.ensure(schema, Jsons.jsonNode(config));
+  }
+
+}

airbyte-config/db-persistence/src/main/java/io.airbyte.config.dbPersistence/PostgresConfigPersistence.java

@@ -0,0 +1,71 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.config.dbPersistence;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.config.ConfigSchema;
+import io.airbyte.db.Databases;
+import io.airbyte.validation.json.JsonSchemaValidator;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.UUID;
+
+public class PostgresConfigPersistence extends DatabaseConfigPersistence {
+
+  public PostgresConfigPersistence(String username, String password, String connectionString) {
+    this(username, password, connectionString, new JsonSchemaValidator());
+  }
+
+  public PostgresConfigPersistence(JsonNode config) {
+    this(config, new JsonSchemaValidator());
+  }
+
+  public PostgresConfigPersistence(JsonNode config, JsonSchemaValidator validator) {
+    this(config.get("username").asText(), config.get("password").asText(), Databases.getPostgresJdbcUrl(config), validator);
+  }
+
+  public PostgresConfigPersistence(String username, String password, String connectionString, JsonSchemaValidator validator) {
+    super(Databases.createJdbcDatabase(username, password, connectionString, Databases.POSTGRES_DRIVER), validator);
+  }
+
+  @Override
+  // Create a table named CONFIG with three columns: CONFIG_ID (PK UUID/string), CONFIG_TYPE (string),
+  // CONFIG_DATA (JSON/string)
+  public void Setup() throws SQLException {
+    database.execute("CREATE TABLE CONFIG (CONFIG_ID UUID PRIMARY KEY, CONFIG_TYPE VARCHAR(32) NOT NULL, CONFIG_DATA JSONB NOT NULL)");
+  }
+
+  @Override
+  protected PreparedStatement writeConfigQuery(Connection conn, ConfigSchema configType, UUID configId, String data) throws SQLException {
+    var result = conn.prepareStatement(
+        "INSERT INTO CONFIG (CONFIG_ID, CONFIG_TYPE, CONFIG_DATA) VALUES (?, ?, CAST(? as jsonb)) ON CONFLICT (CONFIG_ID) DO UPDATE SET CONFIG_DATA = EXCLUDED.CONFIG_DATA");
+    result.setObject(1, configId);
+    result.setString(2, configType.toString());
+    result.setString(3, data);
+    return result;
+  }
+
+}

airbyte-config/db-persistence/src/test/java/io/airbyte/config/dbPersistence/PostgresConfigPersistenceTest.java

@@ -0,0 +1,139 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.config.dbPersistence;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import com.google.common.collect.Sets;
+import io.airbyte.config.ConfigSchema;
+import io.airbyte.config.StandardSourceDefinition;
+import io.airbyte.config.StandardSync;
+import io.airbyte.config.persistence.ConfigNotFoundException;
+import io.airbyte.validation.json.JsonSchemaValidator;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.UUID;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.PostgreSQLContainer;
+
+public class PostgresConfigPersistenceTest {
+
+  public static final UUID UUID_1 = new UUID(0, 1);
+  public static final StandardSourceDefinition SOURCE_1 = new StandardSourceDefinition();
+
+  static {
+    SOURCE_1.withSourceDefinitionId(UUID_1)
+        .withName("apache storm");
+  }
+
+  public static final UUID UUID_2 = new UUID(0, 2);
+  public static final StandardSourceDefinition SOURCE_2 = new StandardSourceDefinition();
+
+  static {
+    SOURCE_2.withSourceDefinitionId(UUID_2)
+        .withName("apache storm");
+  }
+
+  private JsonSchemaValidator schemaValidator;
+
+  private PostgresConfigPersistence configPersistence;
+
+  private static final String IMAGE_NAME = "postgres:13-alpine";
+  private PostgreSQLContainer<?> db;
+
+  @BeforeEach
+  void setUp() throws SQLException {
+    schemaValidator = mock(JsonSchemaValidator.class);
+    db = new PostgreSQLContainer<>(IMAGE_NAME);
+    db.start();
+    configPersistence = new PostgresConfigPersistence(db.getUsername(), db.getPassword(), db.getJdbcUrl(), schemaValidator);
+    configPersistence.Setup();
+  }
+
+  @AfterEach
+  void tearDown() {
+    db.stop();
+    db.close();
+  }
+
+  @Test
+  void testReadWriteConfig() throws IOException, JsonValidationException, ConfigNotFoundException {
+    configPersistence.writeConfig(ConfigSchema.STANDARD_SOURCE_DEFINITION, UUID_1, SOURCE_1);
+
+    assertEquals(
+        SOURCE_1,
+        configPersistence.getConfig(
+            ConfigSchema.STANDARD_SOURCE_DEFINITION,
+            UUID_1,
+            StandardSourceDefinition.class));
+  }
+
+  @Test
+  void testListConfigs() throws JsonValidationException, IOException {
+    configPersistence.writeConfig(ConfigSchema.STANDARD_SOURCE_DEFINITION, UUID_1, SOURCE_1);
+    configPersistence.writeConfig(ConfigSchema.STANDARD_SOURCE_DEFINITION, UUID_2, SOURCE_2);
+
+    assertEquals(
+        Sets.newHashSet(SOURCE_1, SOURCE_2),
+        Sets.newHashSet(configPersistence.listConfigs(ConfigSchema.STANDARD_SOURCE_DEFINITION, StandardSourceDefinition.class)));
+  }
+
+  @Test
+  void writeConfigWithJsonSchemaRef() throws JsonValidationException, IOException, ConfigNotFoundException {
+    final StandardSync standardSync = new StandardSync()
+        .withName("sync")
+        .withPrefix("sync")
+        .withConnectionId(UUID_1)
+        .withSourceId(UUID.randomUUID())
+        .withDestinationId(UUID.randomUUID())
+        .withOperationIds(List.of(UUID.randomUUID()));
+
+    configPersistence.writeConfig(ConfigSchema.STANDARD_SYNC, UUID_1, standardSync);
+
+    assertEquals(
+        standardSync,
+        configPersistence.getConfig(ConfigSchema.STANDARD_SYNC, UUID_1, StandardSync.class));
+  }
+
+  @Test
+  void writeConfigInvalidConfig() throws JsonValidationException {
+    StandardSourceDefinition standardSourceDefinition = SOURCE_1.withName(null);
+
+    doThrow(new JsonValidationException("error")).when(schemaValidator).ensure(any(), any());
+
+    assertThrows(JsonValidationException.class, () -> configPersistence.writeConfig(
+        ConfigSchema.STANDARD_SOURCE_DEFINITION,
+        UUID_1,
+        standardSourceDefinition));
+  }
+
+}

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigNotFoundException.java

@@ -25,16 +25,17 @@
 package io.airbyte.config.persistence;
 
 import io.airbyte.config.ConfigSchema;
+import java.util.UUID;
 
 public class ConfigNotFoundException extends Exception {
 
   private ConfigSchema type;
   private final String configId;
 
-  public ConfigNotFoundException(ConfigSchema type, String configId) {
-    super(String.format("config type: %s id: %s", type, configId));
+  public ConfigNotFoundException(ConfigSchema type, UUID configId) {
+    super(String.format("config type: %s id: %s", type, configId.toString()));
     this.type = type;
-    this.configId = configId;
+    this.configId = configId.toString();
   }
 
   public ConfigSchema getType() {

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistence.java

@@ -28,13 +28,14 @@
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.util.List;
+import java.util.UUID;
 
 public interface ConfigPersistence {
 
-  <T> T getConfig(ConfigSchema configType, String configId, Class<T> clazz) throws ConfigNotFoundException, JsonValidationException, IOException;
+  <T> T getConfig(ConfigSchema configType, UUID configId, Class<T> clazz) throws ConfigNotFoundException, JsonValidationException, IOException;
 
   <T> List<T> listConfigs(ConfigSchema configType, Class<T> clazz) throws JsonValidationException, IOException;
 
-  <T> void writeConfig(ConfigSchema configType, String configId, T config) throws JsonValidationException, IOException;
+  <T> void writeConfig(ConfigSchema configType, UUID configId, T config) throws JsonValidationException, IOException;
 
 }