diff --git a/integrations/Gemfile.lock b/integrations/Gemfile.lock index ba46adbee..6b277c830 100644 --- a/integrations/Gemfile.lock +++ b/integrations/Gemfile.lock @@ -7,7 +7,7 @@ GIT PATH remote: . specs: - multiwoven-integrations (0.35.0) + multiwoven-integrations (0.38.0) MailchimpMarketing activesupport async-websocket diff --git a/integrations/lib/multiwoven/integrations.rb b/integrations/lib/multiwoven/integrations.rb index 4c49cac21..012614ed1 100644 --- a/integrations/lib/multiwoven/integrations.rb +++ b/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" diff --git a/integrations/lib/multiwoven/integrations/destination/mysql/client.rb b/integrations/lib/multiwoven/integrations/destination/mysql/client.rb new file mode 100644 index 000000000..a4532535d --- /dev/null +++ b/integrations/lib/multiwoven/integrations/destination/mysql/client.rb @@ -0,0 +1,137 @@ +# 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 + db = 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 + ensure + db&.disconnect + end + + def discover(connection_config) + connection_config = connection_config.with_indifferent_access + db = create_connection(connection_config) + query = "SELECT table_name, column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = ? + ORDER BY table_name, ordinal_position;" + records = db.fetch(query, connection_config[:database]).all.map { |row| row.transform_keys { |k| k.to_s.downcase.to_sym } } + catalog = Catalog.new(streams: create_streams(records)) + catalog.to_multiwoven_message + rescue StandardError => e + handle_exception(e, { + context: "MYSQL:DISCOVER:EXCEPTION", + type: "error" + }) + ensure + db&.disconnect + 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) + + create_table_if_not_exists(db, table_name, sync_config.stream.json_schema, primary_key) + + 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 action = #{action} table_name = #{table_name} primary_key = #{primary_key} 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", "#{action} #{table_name}", "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", "#{action} #{table_name}", 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 + }) + ensure + db&.disconnect + 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_table_if_not_exists(db, table_name, json_schema, primary_key) + properties = json_schema&.[]("properties") + return unless properties&.any? + + columns = properties.map do |name, prop| + "`#{name}` #{mysql_type_mapping(prop["type"])}" + end + columns << "PRIMARY KEY (`#{primary_key}`)" if primary_key.present? + + db.run("CREATE TABLE IF NOT EXISTS `#{table_name}` (#{columns.join(', ')})") + rescue Sequel::DatabaseError => e + logger.warn("MYSQL:CREATE_TABLE:ERROR #{e.message}") + end + + def mysql_type_mapping(data_type) + case data_type + when "string" then "VARCHAR(255)" + when "integer" then "INT" + when "number" then "DECIMAL(15,6)" + when "boolean" then "TINYINT(1)" + else "TEXT" + end + 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 do |entry| + table_name = entry[:table_name] + column_data = { + column_name: entry[:column_name], + data_type: entry[:data_type], + is_nullable: entry[:is_nullable] == "YES" + } + result[table_name] ||= { tablename: table_name, columns: [] } + result[table_name][:columns] << column_data + end + result + end + end + end +end diff --git a/integrations/lib/multiwoven/integrations/destination/mysql/config/meta.json b/integrations/lib/multiwoven/integrations/destination/mysql/config/meta.json new file mode 100644 index 000000000..035eb62d8 --- /dev/null +++ b/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"] + } +} diff --git a/integrations/lib/multiwoven/integrations/destination/mysql/config/spec.json b/integrations/lib/multiwoven/integrations/destination/mysql/config/spec.json new file mode 100644 index 000000000..d225be6e9 --- /dev/null +++ b/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 + } + } + } +} diff --git a/integrations/lib/multiwoven/integrations/destination/mysql/icon.svg b/integrations/lib/multiwoven/integrations/destination/mysql/icon.svg new file mode 100644 index 000000000..54abbe2a2 --- /dev/null +++ b/integrations/lib/multiwoven/integrations/destination/mysql/icon.svg @@ -0,0 +1,6 @@ + + + + + MySQL + diff --git a/integrations/lib/multiwoven/integrations/destination/postgresql/client.rb b/integrations/lib/multiwoven/integrations/destination/postgresql/client.rb index cfed42c05..8d70741a5 100644 --- a/integrations/lib/multiwoven/integrations/destination/postgresql/client.rb +++ b/integrations/lib/multiwoven/integrations/destination/postgresql/client.rb @@ -54,6 +54,8 @@ def write(sync_config, records, action = "destination_insert") primary_key = sync_config.model.primary_key db = create_connection(connection_config) + create_table_if_not_exists(db, raw_table, sync_config.stream.json_schema, primary_key) + write_success = 0 write_failure = 0 log_message_array = [] @@ -177,6 +179,30 @@ def group_by_table(records) end end + def create_table_if_not_exists(db, table_name, json_schema, primary_key) + properties = json_schema&.[]("properties") + return unless properties&.any? + + columns = properties.map do |name, prop| + "#{quote_ident(name)} #{pg_type_mapping(prop["type"])}" + end + columns << "PRIMARY KEY (#{quote_ident(primary_key)})" if primary_key.present? + + db.exec("CREATE TABLE IF NOT EXISTS #{quote_ident(table_name)} (#{columns.join(', ')})") + rescue PG::Error => e + logger.warn("POSTGRESQL:CREATE_TABLE:ERROR #{e.message}") + end + + def pg_type_mapping(data_type) + case data_type + when "string" then "VARCHAR(255)" + when "integer" then "INTEGER" + when "number" then "DECIMAL" + when "boolean" then "BOOLEAN" + else "TEXT" + end + end + def qualify_table(schema, table) return table if schema.blank? || schema == "public" diff --git a/integrations/lib/multiwoven/integrations/rollout.rb b/integrations/lib/multiwoven/integrations/rollout.rb index ae1dd8962..1abdd6592 100644 --- a/integrations/lib/multiwoven/integrations/rollout.rb +++ b/integrations/lib/multiwoven/integrations/rollout.rb @@ -2,7 +2,7 @@ module Multiwoven module Integrations - VERSION = "0.35.0" + VERSION = "0.39.0" 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 diff --git a/integrations/lib/multiwoven/integrations/source/mysql/client.rb b/integrations/lib/multiwoven/integrations/source/mysql/client.rb new file mode 100644 index 000000000..48f78b3e5 --- /dev/null +++ b/integrations/lib/multiwoven/integrations/source/mysql/client.rb @@ -0,0 +1,97 @@ +# 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 + db = 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 + ensure + db&.disconnect + end + + def discover(connection_config) + connection_config = connection_config.with_indifferent_access + db = create_connection(connection_config) + query = "SELECT table_name, column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = ? ORDER BY table_name, ordinal_position;" + results = query_execution(db, query, connection_config[:database]) + catalog = Catalog.new(streams: create_streams(results)) + catalog.to_multiwoven_message + rescue StandardError => e + handle_exception(e, { + context: "MYSQL:DISCOVER:EXCEPTION", + type: "error" + }) + ensure + db&.disconnect + 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 + }) + ensure + db&.disconnect + 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, *params) + db.fetch(query, *params).all.map { |row| row.transform_keys { |k| k.to_s.downcase.to_sym } } + 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 do |entry| + table_name = entry[:table_name] + column_data = { + column_name: entry[:column_name], + data_type: entry[:data_type], + is_nullable: entry[:is_nullable] == "YES" + } + result[table_name] ||= { tablename: table_name, columns: [] } + result[table_name][:columns] << column_data + end + result + end + end + end +end diff --git a/integrations/lib/multiwoven/integrations/source/mysql/config/meta.json b/integrations/lib/multiwoven/integrations/source/mysql/config/meta.json new file mode 100644 index 000000000..295eda950 --- /dev/null +++ b/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"] + } +} diff --git a/integrations/lib/multiwoven/integrations/source/mysql/config/spec.json b/integrations/lib/multiwoven/integrations/source/mysql/config/spec.json new file mode 100644 index 000000000..2c350bfe5 --- /dev/null +++ b/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 + } + } + } +} diff --git a/integrations/lib/multiwoven/integrations/source/mysql/icon.svg b/integrations/lib/multiwoven/integrations/source/mysql/icon.svg new file mode 100644 index 000000000..54abbe2a2 --- /dev/null +++ b/integrations/lib/multiwoven/integrations/source/mysql/icon.svg @@ -0,0 +1,6 @@ + + + + + MySQL + diff --git a/integrations/multiwoven-integrations.gemspec b/integrations/multiwoven-integrations.gemspec index e4e113bd6..a1b2deb45 100644 --- a/integrations/multiwoven-integrations.gemspec +++ b/integrations/multiwoven-integrations.gemspec @@ -24,9 +24,13 @@ Gem::Specification.new do |spec| # Specify which files should be added to the gem when it is released. # The `git ls-files -z` loads the files in the RubyGem that have been added into git. spec.files = Dir.chdir(__dir__) do - `git ls-files -z`.split("\x0").reject do |f| - (File.expand_path(f) == __FILE__) || - f.start_with?(*%w[bin/ test/ spec/ features/ .git .circleci appveyor Gemfile]) + if system("git rev-parse --is-inside-work-tree 2>/dev/null", out: File::NULL) + `git ls-files -z`.split("\x0").reject do |f| + (File.expand_path(f) == __FILE__) || + f.start_with?(*%w[bin/ test/ spec/ features/ .git .circleci appveyor Gemfile]) + end + else + Dir.glob("lib/**/*") end end spec.bindir = "exe" diff --git a/integrations/spec/multiwoven/integrations/destination/mysql/client_spec.rb b/integrations/spec/multiwoven/integrations/destination/mysql/client_spec.rb new file mode 100644 index 000000000..cb7861a8c --- /dev/null +++ b/integrations/spec/multiwoven/integrations/destination/mysql/client_spec.rb @@ -0,0 +1,158 @@ +# frozen_string_literal: true + +RSpec.describe Multiwoven::Integrations::Destination::Mysql::Client do + include WebMock::API + + before(:each) do + WebMock.disable_net_connect!(allow_localhost: true) + end + + let(:client) { described_class.new } + let(:connection_config) do + { + host: "127.0.0.1", + port: "3306", + username: "test_user", + password: ENV["MYSQL_PASSWORD"], + database: "test_database" + } + end + let(:sync_config_json) do + { + source: { + name: "Sample Source Connector", + type: "source", + connection_specification: { + private_api_key: "test_api_key" + } + }, + destination: { + name: "Mysql", + type: "destination", + connection_specification: connection_config + }, + model: { + name: "ExampleModel", + query: "SELECT col1, col2, col3 FROM test_table", + query_type: "raw_sql", + primary_key: "col1" + }, + sync_mode: "incremental", + destination_sync_mode: "insert", + stream: { + name: "test_table", + action: "create", + json_schema: { "field1": "type1" }, + supported_sync_modes: %w[full_refresh incremental] + } + } + end + + let(:sequel_client) { instance_double(Sequel::Database) } + + before(:each) do + allow(sequel_client).to receive(:disconnect) + end + + describe "#check_connection" do + context "when the connection is successful" do + it "returns a succeeded connection status" do + allow_any_instance_of(Multiwoven::Integrations::Destination::Mysql::Client).to receive(:create_connection).and_return(sequel_client) + message = client.check_connection(sync_config_json[:destination][:connection_specification]) + result = message.connection_status + expect(result.status).to eq("succeeded") + expect(result.message).to be_nil + end + end + + context "when the connection fails" do + it "returns a failed connection status with an error message" do + allow_any_instance_of(Multiwoven::Integrations::Destination::Mysql::Client).to receive(:create_connection).and_raise(StandardError, "Connection failed") + message = client.check_connection(sync_config_json[:destination][:connection_specification]) + result = message.connection_status + expect(result.status).to eq("failed") + expect(result.message).to include("Connection failed") + end + end + end + + describe "#discover" do + it "discovers schema successfully" do + dataset = double("Sequel::Dataset") + rows = [ + { table_name: "test_table", column_name: "col1", data_type: "int", is_nullable: "YES" }, + { table_name: "test_table", column_name: "col2", data_type: "varchar", is_nullable: "YES" }, + { table_name: "test_table", column_name: "col3", data_type: "float", is_nullable: "YES" } + ] + allow(dataset).to receive(:all).and_return(rows) + allow(sequel_client).to receive(:fetch).and_return(dataset) + allow(client).to receive(:create_connection).and_return(sequel_client) + + message = client.discover(sync_config_json[:destination][:connection_specification]) + expect(message.catalog).to be_an(Multiwoven::Integrations::Protocol::Catalog) + first_stream = message.catalog.streams.first + expect(first_stream).to be_a(Multiwoven::Integrations::Protocol::Stream) + expect(first_stream.name).to eq("test_table") + expect(first_stream.json_schema).to be_an(Hash) + expect(first_stream.json_schema["type"]).to eq("object") + expect(first_stream.json_schema["properties"]).to eq({ "col1" => { "type" => "string" }, "col2" => { "type" => "string" }, "col3" => { "type" => "string" } }) + end + end + + describe "#write" do + context "when the write operation is successful" do + it "increments the success count" do + sync_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json( + sync_config_json.to_json + ) + records = [ + { "col1" => 400, "col2" => 4.4, "col3" => "Fourth" }, + { "col1" => 500, "col2" => 5.5, "col3" => "Fifth" }, + { "col1" => 600, "col2" => 6.6, "col3" => "Sixth" } + ] + allow(client).to receive(:create_connection).and_return(sequel_client) + allow(sequel_client).to receive(:run).and_return(nil) + response = client.write(sync_config, records) + expect(response.tracking.success).to eq(records.size) + expect(response.tracking.failed).to eq(0) + log_message = response.tracking.logs.first + expect(log_message).to be_a(Multiwoven::Integrations::Protocol::LogMessage) + expect(log_message.level).to eql("info") + + expect(log_message.message).to include("request") + expect(log_message.message).to include("response") + end + end + + context "when the write operation fails" do + it "increments the failure count" do + sync_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json( + sync_config_json.to_json + ) + records = [ + { "col1" => 400, "col2" => 4.4, "col3" => "Fourth" }, + { "col1" => 500, "col2" => 5.5, "col3" => "Fifth" }, + { "col1" => 600, "col2" => 6.6, "col3" => "Sixth" } + ] + allow(client).to receive(:create_connection).and_return(sequel_client) + allow(sequel_client).to receive(:run).and_raise(StandardError) + response = client.write(sync_config, records) + expect(response.tracking.failed).to eq(records.size) + expect(response.tracking.success).to eq(0) + log_message = response.tracking.logs.first + expect(log_message).to be_a(Multiwoven::Integrations::Protocol::LogMessage) + expect(log_message.level).to eql("error") + + expect(log_message.message).to include("request") + expect(log_message.message).to include("response") + end + end + end + + describe "#meta_data" do + it "client class_name and meta name is same" do + meta_name = client.class.to_s.split("::")[-2] + expect(client.send(:meta_data)[:data][:name]).to eq(meta_name) + end + end +end diff --git a/integrations/spec/multiwoven/integrations/source/mysql/client_spec.rb b/integrations/spec/multiwoven/integrations/source/mysql/client_spec.rb new file mode 100644 index 000000000..f1a902327 --- /dev/null +++ b/integrations/spec/multiwoven/integrations/source/mysql/client_spec.rb @@ -0,0 +1,179 @@ +# frozen_string_literal: true + +RSpec.describe Multiwoven::Integrations::Source::Mysql::Client do + let(:client) { Multiwoven::Integrations::Source::Mysql::Client.new } + let(:sync_config) do + { + "source": { + "name": "MySQLConnector", + "type": "source", + "connection_specification": { + "host": "127.0.0.1", + "port": "3306", + "username": "test_user", + "password": ENV["MYSQL_PASSWORD"], + "database": "test_database" + } + }, + "destination": { + "name": "DestinationConnectorName", + "type": "destination", + "connection_specification": { + "example_destination_key": "example_destination_value" + } + }, + "model": { + "name": "MySQL Model", + "query": "SELECT col1, col2, col3 FROM test_table", + "query_type": "raw_sql", + "primary_key": "id" + }, + "stream": { + "name": "example_stream", "action": "create", + "json_schema": { "field1": "type1" }, + "supported_sync_modes": %w[full_refresh incremental], + "source_defined_cursor": true, + "default_cursor_field": ["field1"], + "source_defined_primary_key": [["field1"], ["field2"]], + "namespace": "exampleNamespace", + "url": "https://api.example.com/data", + "method": "GET" + }, + "sync_mode": "full_refresh", + "cursor_field": "timestamp", + "destination_sync_mode": "upsert", + "sync_id": "1" + } + end + + let(:sequel_client) { instance_double(Sequel::Database) } + + before(:each) do + allow(sequel_client).to receive(:disconnect) + end + + describe "#check_connection" do + context "when the connection is successful" do + it "returns a succeeded connection status" do + allow_any_instance_of(Multiwoven::Integrations::Source::Mysql::Client).to receive(:create_connection).and_return(sequel_client) + message = client.check_connection(sync_config[:source][:connection_specification]) + result = message.connection_status + expect(result.status).to eq("succeeded") + expect(result.message).to be_nil + end + end + + context "when the connection fails" do + it "returns a failed connection status with an error message" do + allow_any_instance_of(Multiwoven::Integrations::Source::Mysql::Client).to receive(:create_connection).and_raise(StandardError, "Connection failed") + message = client.check_connection(sync_config[:source][:connection_specification]) + result = message.connection_status + expect(result.status).to eq("failed") + expect(result.message).to include("Connection failed") + end + end + end + + describe "#read" do + it "reads records successfully" do + s_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json(sync_config.to_json) + dataset = double("Sequel::Dataset") + allow(dataset).to receive(:all).and_return( + [ + { col1: 1, col2: "First Row Text", col3: "First Row Additional Text" }, + { col1: 1, col2: "Second Row Text", col3: "Second Row Additional Text" }, + { col1: 1, col2: "Third Row Text", col3: "Third Row Additional Text" } + ] + ) + allow(sequel_client).to receive(:fetch).and_return(dataset) + allow(client).to receive(:create_connection).and_return(sequel_client) + records = client.read(s_config) + expect(records).to be_an(Array) + expect(records).not_to be_empty + expect(records.first).to be_a(Multiwoven::Integrations::Protocol::MultiwovenMessage) + end + + it "reads records successfully with limit" do + s_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json(sync_config.to_json) + s_config.limit = 100 + s_config.offset = 1 + dataset = double("Sequel::Dataset") + allow(dataset).to receive(:all).and_return( + [ + { col1: 1, col2: "First Row Text", col3: "First Row Additional Text" }, + { col1: 1, col2: "Second Row Text", col3: "Second Row Additional Text" }, + { col1: 1, col2: "Third Row Text", col3: "Third Row Additional Text" } + ] + ) + expect(sequel_client).to receive(:fetch).with(a_string_including("LIMIT 100").and(a_string_including("OFFSET 1"))).and_return(dataset) + allow(client).to receive(:create_connection).and_return(sequel_client) + records = client.read(s_config) + expect(records).to be_an(Array) + expect(records).not_to be_empty + expect(records.first).to be_a(Multiwoven::Integrations::Protocol::MultiwovenMessage) + end + + it "read records failure" do + s_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json(sync_config.to_json) + s_config.sync_run_id = "2" + allow(client).to receive(:create_connection).and_raise(StandardError, "test error") + expect(client).to receive(:handle_exception).with( + an_instance_of(StandardError), { + context: "MYSQL:READ:EXCEPTION", + type: "error", + sync_id: "1", + sync_run_id: "2" + } + ) + client.read(s_config) + end + end + + describe "#discover" do + it "discovers schema successfully" do + dataset = double("Sequel::Dataset") + allow(dataset).to receive(:all).and_return( + [ + { table_name: "test_table", column_name: "col1", data_type: "int", is_nullable: "YES" }, + { table_name: "test_table", column_name: "col2", data_type: "varchar", is_nullable: "YES" }, + { table_name: "test_table", column_name: "col3", data_type: "float", is_nullable: "YES" } + ] + ) + allow(sequel_client).to receive(:fetch).and_return(dataset) + allow(client).to receive(:create_connection).and_return(sequel_client) + + message = client.discover(sync_config[:source][:connection_specification]) + expect(message.catalog).to be_an(Multiwoven::Integrations::Protocol::Catalog) + first_stream = message.catalog.streams.first + expect(first_stream).to be_a(Multiwoven::Integrations::Protocol::Stream) + expect(first_stream.name).to eq("test_table") + expect(first_stream.json_schema).to be_an(Hash) + expect(first_stream.json_schema["type"]).to eq("object") + expect(first_stream.json_schema["properties"]).to eq({ "col1" => { "type" => "string" }, "col2" => { "type" => "string" }, "col3" => { "type" => "string" } }) + end + + it "discover schema failure" do + allow(client).to receive(:create_connection).and_raise(StandardError, "test error") + expect(client).to receive(:handle_exception).with( + an_instance_of(StandardError), { + context: "MYSQL:DISCOVER:EXCEPTION", + type: "error" + } + ) + client.discover(sync_config[:source][:connection_specification]) + end + end + + describe "#meta_data" do + it "client class_name and meta name is same" do + meta_name = client.class.to_s.split("::")[-2] + expect(client.send(:meta_data)[:data][:name]).to eq(meta_name) + end + end + + describe "method definition" do + it "defines a private #query method" do + expect(described_class.private_instance_methods).to include(:query) + end + end +end diff --git a/server/Gemfile b/server/Gemfile index bb94e30ae..90d04b28c 100644 --- a/server/Gemfile +++ b/server/Gemfile @@ -14,7 +14,7 @@ gem "interactor", "~> 3.0" gem "ruby-odbc", git: "https://github.com/Multiwoven/ruby-odbc.git" -gem "multiwoven-integrations", "~> 0.34.6" +gem "multiwoven-integrations", "~> 0.39.0" gem "temporal-ruby", github: "coinbase/temporal-ruby" diff --git a/server/Gemfile.lock b/server/Gemfile.lock index 8b7d05ce5..213a70d57 100644 --- a/server/Gemfile.lock +++ b/server/Gemfile.lock @@ -1979,7 +1979,7 @@ GEM multi_xml (0.7.2) bigdecimal (~> 3.1) multipart-post (2.4.1) - multiwoven-integrations (0.34.6) + multiwoven-integrations (0.38.0) MailchimpMarketing activesupport async-websocket @@ -2339,7 +2339,7 @@ DEPENDENCIES kaminari letter_opener liquid - multiwoven-integrations (~> 0.34.6) + multiwoven-integrations (~> 0.38.0) mysql2 newrelic_rpm parallel