feat(CE): add mysql source and destination connectors

integrations/lib/multiwoven/integrations.rb

@@ -80,6 +80,7 @@
 require_relative "integrations/source/clickhouse/client"
 require_relative "integrations/source/amazon_s3/client"
 require_relative "integrations/source/maria_db/client"
+require_relative "integrations/source/mysql/client"
 require_relative "integrations/source/oracle_db/client"
 require_relative "integrations/source/databrics_model/client"
 require_relative "integrations/source/aws_sagemaker_model/client"
@@ -116,6 +117,7 @@
 require_relative "integrations/destination/http/client"
 require_relative "integrations/destination/iterable/client"
 require_relative "integrations/destination/maria_db/client"
+require_relative "integrations/destination/mysql/client"
 require_relative "integrations/destination/databricks_lakehouse/client"
 require_relative "integrations/destination/oracle_db/client"
 require_relative "integrations/destination/microsoft_excel/client"

integrations/lib/multiwoven/integrations/destination/mysql/client.rb

@@ -0,0 +1,113 @@
+# frozen_string_literal: true
+
+module Multiwoven::Integrations::Destination
+  module Mysql
+    include Multiwoven::Integrations::Core
+    class Client < DestinationConnector
+      def check_connection(connection_config)
+        connection_config = connection_config.with_indifferent_access
+        create_connection(connection_config)
+        ConnectionStatus.new(status: ConnectionStatusType["succeeded"]).to_multiwoven_message
+      rescue StandardError => e
+        ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.message).to_multiwoven_message
+      end
+
+      def discover(connection_config)
+        connection_config = connection_config.with_indifferent_access
+        query = "SELECT table_name, column_name, data_type, is_nullable
+                 FROM information_schema.columns
+                 WHERE table_schema = '#{connection_config[:database]}'
+                 ORDER BY table_name, ordinal_position;"
+
+        db = create_connection(connection_config)
+        records = db.fetch(query) do |result|
+          result.map do |row|
+            row
+          end
+        end
+        catalog = Catalog.new(streams: create_streams(records))
+        catalog.to_multiwoven_message
+      rescue StandardError => e
+        handle_exception(e, {
+                           context: "MYSQL:DISCOVER:EXCEPTION",
+                           type: "error"
+                         })
+      end
+
+      def write(sync_config, records, action = "destination_insert")
+        connection_config = sync_config.destination.connection_specification.with_indifferent_access
+        table_name = sync_config.stream.name
+        primary_key = sync_config.model.primary_key
+        db = create_connection(connection_config)
+
+        log_message_array = []
+        write_success = 0
+        write_failure = 0
+
+        records.each do |record|
+          query = Multiwoven::Integrations::Core::QueryBuilder.perform(action, table_name, record, primary_key)
+          logger.debug("MYSQL:WRITE:QUERY query = #{query} sync_id = #{sync_config.sync_id} sync_run_id = #{sync_config.sync_run_id}")
+          begin
+            db.run(query)
+            write_success += 1
+            log_message_array << log_request_response("info", query, "Successful")
+          rescue StandardError => e
+            handle_exception(e, {
+                               context: "MYSQL:RECORD:WRITE:EXCEPTION",
+                               type: "error",
+                               sync_id: sync_config.sync_id,
+                               sync_run_id: sync_config.sync_run_id
+                             })
+            write_failure += 1
+            log_message_array << log_request_response("error", query, e.message)
+          end
+        end
+        tracking_message(write_success, write_failure, log_message_array)
+      rescue StandardError => e
+        handle_exception(e, {
+                           context: "MYSQL:WRITE:EXCEPTION",
+                           type: "error",
+                           sync_id: sync_config.sync_id,
+                           sync_run_id: sync_config.sync_run_id
+                         })
+      end
+
+      private
+
+      def create_connection(connection_config)
+        Sequel.connect(
+          adapter: "mysql2",
+          host: connection_config[:host],
+          port: connection_config[:port],
+          user: connection_config[:username],
+          password: connection_config[:password],
+          database: connection_config[:database]
+        )
+      end
+
+      def create_streams(records)
+        group_by_table(records).map do |_, r|
+          Multiwoven::Integrations::Protocol::Stream.new(name: r[:tablename], action: StreamAction["fetch"], json_schema: convert_to_json_schema(r[:columns]))
+        end
+      end
+
+      def group_by_table(records)
+        result = {}
+        records.each_with_index do |entry, index|
+          table_name = entry[:table_name]
+          column_data = {
+            column_name: entry[:column_name],
+            data_type: entry[:data_type],
+            is_nullable: entry[:is_nullable] == "YES"
+          }
+          result[index] ||= {}
+          result[index][:tablename] = table_name
+          result[index][:columns] = [column_data]
+        end
+        result.values.group_by { |entry| entry[:tablename] }.transform_values do |entries|
+          { tablename: entries.first[:tablename], columns: entries.flat_map { |entry| entry[:columns] } }
+        end
+      end
+    end
+  end
+end

integrations/lib/multiwoven/integrations/destination/mysql/config/meta.json

@@ -0,0 +1,15 @@
+{
+  "data": {
+    "name": "Mysql",
+    "title": "MySQL",
+    "connector_type": "destination",
+    "category": "Database",
+    "documentation_url": "https://docs.multiwoven.com/guides/destinations/mysql",
+    "github_issue_label": "destination-mysql",
+    "icon": "icon.svg",
+    "license": "MIT",
+    "release_stage": "alpha",
+    "support_level": "community",
+    "tags": ["language:ruby", "multiwoven"]
+  }
+}

integrations/lib/multiwoven/integrations/destination/mysql/config/spec.json

@@ -0,0 +1,48 @@
+{
+  "documentation_url": "https://docs.multiwoven.com/guides/destinations/mysql",
+  "stream_type": "dynamic",
+  "connector_query_type": "raw_sql",
+  "connection_specification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "MySQL",
+    "type": "object",
+    "required": ["host", "port", "username", "password", "database"],
+    "properties": {
+      "host": {
+        "description": "The hostname or IP address of the server where the MySQL database is hosted.",
+        "examples": ["localhost"],
+        "type": "string",
+        "title": "Host",
+        "order": 0
+      },
+      "port": {
+        "description": "The port number on which the MySQL server is listening for connections.",
+        "examples": ["3306"],
+        "type": "string",
+        "title": "Port",
+        "order": 1
+      },
+      "username": {
+        "description": "The username used to authenticate and connect to the MySQL database.",
+        "examples": ["root"],
+        "type": "string",
+        "title": "Username",
+        "order": 2
+      },
+      "password": {
+        "description": "The password corresponding to the username used for authentication.",
+        "type": "string",
+        "multiwoven_secret": true,
+        "title": "Password",
+        "order": 3
+      },
+      "database": {
+        "description": "The name of the specific database within the MySQL server to connect to.",
+        "examples": ["mydatabase"],
+        "type": "string",
+        "title": "Database",
+        "order": 4
+      }
+    }
+  }
+}

integrations/lib/multiwoven/integrations/rollout.rb

@@ -2,7 +2,7 @@
 
 module Multiwoven
   module Integrations
-    VERSION = "0.35.0"
+    VERSION = "0.35.1"
 
     ENABLED_SOURCES = %w[
       Snowflake
@@ -15,6 +15,7 @@ module Integrations
       Clickhouse
       AmazonS3
       MariaDB
+      Mysql
       Oracle
       DatabricksModel
       AwsSagemakerModel
@@ -52,6 +53,7 @@ module Integrations
       Http
       Iterable
       MariaDB
+      Mysql
       DatabricksLakehouse
       Oracle
       MicrosoftExcel

integrations/lib/multiwoven/integrations/source/mysql/client.rb

@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+module Multiwoven::Integrations::Source
+  module Mysql
+    include Multiwoven::Integrations::Core
+    class Client < SourceConnector
+      def check_connection(connection_config)
+        connection_config = connection_config.with_indifferent_access
+        create_connection(connection_config)
+        ConnectionStatus.new(status: ConnectionStatusType["succeeded"]).to_multiwoven_message
+      rescue StandardError => e
+        ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.message).to_multiwoven_message
+      end
+
+      def discover(connection_config)
+        connection_config = connection_config.with_indifferent_access
+        query = "SELECT table_name, column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = '#{connection_config[:database]}' ORDER BY table_name, ordinal_position;"
+        db = create_connection(connection_config)
+        results = query_execution(db, query)
+        catalog = Catalog.new(streams: create_streams(results))
+        catalog.to_multiwoven_message
+      rescue StandardError => e
+        handle_exception(e, {
+                           context: "MYSQL:DISCOVER:EXCEPTION",
+                           type: "error"
+                         })
+      end
+
+      def read(sync_config)
+        connection_config = sync_config.source.connection_specification.with_indifferent_access
+        query = sync_config.model.query
+        query = batched_query(query, sync_config.limit, sync_config.offset) unless sync_config.limit.nil? && sync_config.offset.nil?
+        db = create_connection(connection_config)
+        query(db, query)
+      rescue StandardError => e
+        handle_exception(e, {
+                           context: "MYSQL:READ:EXCEPTION",
+                           type: "error",
+                           sync_id: sync_config.sync_id,
+                           sync_run_id: sync_config.sync_run_id
+                         })
+      end
+
+      private
+
+      def create_connection(connection_config)
+        Sequel.connect(
+          adapter: "mysql2",
+          host: connection_config[:host],
+          port: connection_config[:port],
+          user: connection_config[:username],
+          password: connection_config[:password],
+          database: connection_config[:database]
+        )
+      end
+
+      def query_execution(db, query)
+        db.fetch(query).all
+      end
+
+      def create_streams(records)
+        group_by_table(records).map do |_, r|
+          Multiwoven::Integrations::Protocol::Stream.new(name: r[:tablename], action: StreamAction["fetch"], json_schema: convert_to_json_schema(r[:columns]))
+        end
+      end
+
+      def query(db, query)
+        records = []
+        query_execution(db, query).map do |row|
+          records << RecordMessage.new(data: row, emitted_at: Time.now.to_i).to_multiwoven_message
+        end
+        records
+      end
+
+      def group_by_table(records)
+        result = {}
+        records.each_with_index do |entry, index|
+          table_name = entry[:table_name]
+          column_data = {
+            column_name: entry[:column_name],
+            data_type: entry[:data_type],
+            is_nullable: entry[:is_nullable] == "YES"
+          }
+          result[index] ||= {}
+          result[index][:tablename] = table_name
+          result[index][:columns] = [column_data]
+        end
+        result
+      end
+    end
+  end
+end

integrations/lib/multiwoven/integrations/source/mysql/config/meta.json

@@ -0,0 +1,16 @@
+{
+  "data": {
+    "name": "Mysql",
+    "title": "MySQL",
+    "connector_type": "source",
+    "category": "Data Warehouse",
+    "sub_category": "Relational Database",
+    "documentation_url": "https://docs.multiwoven.com/guides/sources/mysql",
+    "github_issue_label": "source-mysql",
+    "icon": "icon.svg",
+    "license": "MIT",
+    "release_stage": "alpha",
+    "support_level": "community",
+    "tags": ["language:ruby", "multiwoven"]
+  }
+}

integrations/lib/multiwoven/integrations/source/mysql/config/spec.json

@@ -0,0 +1,48 @@
+{
+  "documentation_url": "https://docs.multiwoven.com/guides/sources/mysql",
+  "stream_type": "dynamic",
+  "connector_query_type": "raw_sql",
+  "connection_specification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "MySQL",
+    "type": "object",
+    "required": ["host", "port", "username", "password", "database"],
+    "properties": {
+      "host": {
+        "description": "The hostname or IP address of the server where the MySQL database is hosted.",
+        "examples": ["localhost"],
+        "type": "string",
+        "title": "Host",
+        "order": 0
+      },
+      "port": {
+        "description": "The port number on which the MySQL server is listening for connections.",
+        "examples": ["3306"],
+        "type": "string",
+        "title": "Port",
+        "order": 1
+      },
+      "username": {
+        "description": "The username used to authenticate and connect to the MySQL database.",
+        "examples": ["root"],
+        "type": "string",
+        "title": "Username",
+        "order": 2
+      },
+      "password": {
+        "description": "The password corresponding to the username used for authentication.",
+        "type": "string",
+        "multiwoven_secret": true,
+        "title": "Password",
+        "order": 3
+      },
+      "database": {
+        "description": "The name of the specific database within the MySQL server to connect to.",
+        "examples": ["mydatabase"],
+        "type": "string",
+        "title": "Database",
+        "order": 4
+      }
+    }
+  }
+}