Add MySQL source connector

Author: tiakumzukimsong

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_db/client"
 require_relative "integrations/source/oracle_db/client"
 require_relative "integrations/source/databrics_model/client"
 require_relative "integrations/source/aws_sagemaker_model/client"

integrations/lib/multiwoven/integrations/rollout.rb

@@ -34,6 +34,7 @@ module Integrations
       Odoo
       GoogleDrive
       Http
+      MysqlDb
     ].freeze
 
     ENABLED_DESTINATIONS = %w[
@@ -63,6 +64,7 @@ module Integrations
       Qdrant
       PineconeDB
       Odoo
+      MysqlDb
     ].freeze
   end
 end

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

@@ -0,0 +1,113 @@
+# frozen_string_literal: true
+
+require "sequel"
+require "multiwoven/integrations/core/source_connector"
+require "multiwoven/integrations/protocol/protocol"
+require "active_support/core_ext/hash/indifferent_access" 
+
+module Multiwoven
+  module Integrations
+    module Source
+      module MysqlDb
+        class Client < Multiwoven::Integrations::Core::SourceConnector
+
+          
+          def check_connection(connection_config)
+            create_connection(connection_config.with_indifferent_access)
+            ConnectionStatus.new(status: ConnectionStatusType["succeeded"]).to_multiwoven_message
+          rescue => e
+            ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.message).to_multiwoven_message
+          end
+
+          def discover(connection_config)
+            cfg = connection_config.with_indifferent_access
+            db = create_connection(cfg)
+
+            query = <<-SQL
+              SELECT table_name, column_name, data_type, is_nullable
+              FROM information_schema.columns
+              WHERE table_schema = '#{cfg[:database]}'
+              ORDER BY table_name, ordinal_position;
+            SQL
+
+            results = db.fetch(query).all
+            catalog = Catalog.new(streams: build_streams(results))
+            catalog.to_multiwoven_message
+
+          rescue => e
+            handle_exception(e, context: "MYSQL:DISCOVER:EXCEPTION", type: "error")
+          end
+
+          def read(sync_config)
+            cfg = sync_config.source.connection_specification.with_indifferent_access
+            db = create_connection(cfg)
+
+            sql = sync_config.model.query
+            sql = batched_query(sql, sync_config.limit, sync_config.offset) if sync_config.limit || sync_config.offset
+
+            rows = db.fetch(sql)
+            rows.map do |row|
+              RecordMessage.new(data: row, emitted_at: Time.now.to_i).to_multiwoven_message
+            end
+
+          rescue => 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(cfg)
+            Sequel.connect(
+              adapter: "mysql2",
+              host: cfg[:host],
+              port: cfg[:port] || 3306,
+              user: cfg[:username],
+              password: cfg[:password],
+              database: cfg[:database]
+            )
+          end
+
+          def build_streams(records)
+  records.group_by { |r| r[:table_name] || r[:TABLE_NAME] }.map do |table_name, cols|
+    schema = convert_schema(cols)
+    Multiwoven::Integrations::Protocol::Stream.new(
+      name: table_name.to_s,
+      action: StreamAction["fetch"],
+      json_schema: schema
+    )
+  end
+end
+
+def convert_schema(columns)
+  properties = {}
+
+  columns.each do |col|
+    col_name = col[:column_name] || col[:COLUMN_NAME]
+    data_type = col[:data_type] || col[:DATA_TYPE]
+
+    properties[col_name] = { type: map_type(data_type) }
+  end
+
+  { type: "object", properties: properties }
+end
+
+          def map_type(data_type)
+            case data_type
+            when /int/ then "integer"
+            when /char|text|varchar/ then "string"
+            when /date|time|timestamp/ then "string"
+            when /decimal|float|double/ then "number"
+            when /bool/ then "boolean"
+            else "string"
+            end
+          end
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,17 @@
+{
+  "data": {
+    "name": "mysql_db",
+    "title": "MySQL",
+    "connector_type": "source",
+    "version": "0.1.0",
+    "category": "Data Warehouse",
+    "sub_category": "Relational Database",
+    "documentation_url": "https://dev.mysql.com/doc/",
+    "github_issue_label": "source-mysql-db",
+    "icon": "icon.svg",
+    "license": "MIT",
+    "release_stage": "alpha",
+    "support_level": "community",
+    "tags": ["language:ruby", "multiwoven", "database", "sql", "mysql"]
+  }
+}

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

@@ -0,0 +1,49 @@
+{
+  "documentation_url": "https://dev.mysql.com/doc/",
+  "stream_type": "dynamic",
+  "connector_query_type": "raw_sql",
+  "connection_specification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "MySQL Database",
+    "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": ["127.0.0.1", "mysql.yourdomain.com","localhost"],
+        "type": "string",
+        "title": "Host",
+        "order": 0
+      },
+      "port": {
+        "description": "The port number on which the MySQL server is listening for connections.",
+        "examples": [3306],
+        "type": "integer",
+        "title": "Port",
+        "default": 3306,
+        "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": ["test", "analytics", "production_db","mydatabase"],
+        "type": "string",
+        "title": "Database",
+        "order": 4
+      }
+    }
+  }
+}