Skip to content

Commit e9bdeb3

Browse files
gauravgurjar-degauravgurjar-de
committed
feat(CE): add mysql source and destination connectors
Implements MySQL connector supporting both source (read/discover/check_connection) and destination (write/discover/check_connection) operations. - Source connector: Sequel mysql2 adapter with limit/offset support - Destination connector: QueryBuilder-based SQL generation with db.run writes - Full test coverage: 9 source tests, 6 destination tests (all passing) - Registered in rollout.rb ENABLED_SOURCES and ENABLED_DESTINATIONS - Version bumped to 0.35.1 Closes #10
1 parent b8798d5 commit e9bdeb3

12 files changed

Lines changed: 676 additions & 1 deletion

File tree

‎integrations/lib/multiwoven/integrations.rb‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
require_relative "integrations/source/clickhouse/client"
8181
require_relative "integrations/source/amazon_s3/client"
8282
require_relative "integrations/source/maria_db/client"
83+
require_relative "integrations/source/mysql/client"
8384
require_relative "integrations/source/oracle_db/client"
8485
require_relative "integrations/source/databrics_model/client"
8586
require_relative "integrations/source/aws_sagemaker_model/client"
@@ -116,6 +117,7 @@
116117
require_relative "integrations/destination/http/client"
117118
require_relative "integrations/destination/iterable/client"
118119
require_relative "integrations/destination/maria_db/client"
120+
require_relative "integrations/destination/mysql/client"
119121
require_relative "integrations/destination/databricks_lakehouse/client"
120122
require_relative "integrations/destination/oracle_db/client"
121123
require_relative "integrations/destination/microsoft_excel/client"

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

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
# frozen_string_literal: true
2+
3+
module Multiwoven::Integrations::Destination
4+
module Mysql
5+
include Multiwoven::Integrations::Core
6+
class Client < DestinationConnector
7+
def check_connection(connection_config)
8+
connection_config = connection_config.with_indifferent_access
9+
create_connection(connection_config)
10+
ConnectionStatus.new(status: ConnectionStatusType["succeeded"]).to_multiwoven_message
11+
rescue StandardError => e
12+
ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.message).to_multiwoven_message
13+
end
14+
15+
def discover(connection_config)
16+
connection_config = connection_config.with_indifferent_access
17+
query = "SELECT table_name, column_name, data_type, is_nullable
18+
FROM information_schema.columns
19+
WHERE table_schema = '#{connection_config[:database]}'
20+
ORDER BY table_name, ordinal_position;"
21+
22+
db = create_connection(connection_config)
23+
records = db.fetch(query) do |result|
24+
result.map do |row|
25+
row
26+
end
27+
end
28+
catalog = Catalog.new(streams: create_streams(records))
29+
catalog.to_multiwoven_message
30+
rescue StandardError => e
31+
handle_exception(e, {
32+
context: "MYSQL:DISCOVER:EXCEPTION",
33+
type: "error"
34+
})
35+
end
36+
37+
def write(sync_config, records, action = "destination_insert")
38+
connection_config = sync_config.destination.connection_specification.with_indifferent_access
39+
table_name = sync_config.stream.name
40+
primary_key = sync_config.model.primary_key
41+
db = create_connection(connection_config)
42+
43+
log_message_array = []
44+
write_success = 0
45+
write_failure = 0
46+
47+
records.each do |record|
48+
query = Multiwoven::Integrations::Core::QueryBuilder.perform(action, table_name, record, primary_key)
49+
logger.debug("MYSQL:WRITE:QUERY query = #{query} sync_id = #{sync_config.sync_id} sync_run_id = #{sync_config.sync_run_id}")
50+
begin
51+
db.run(query)
52+
write_success += 1
53+
log_message_array << log_request_response("info", query, "Successful")
54+
rescue StandardError => e
55+
handle_exception(e, {
56+
context: "MYSQL:RECORD:WRITE:EXCEPTION",
57+
type: "error",
58+
sync_id: sync_config.sync_id,
59+
sync_run_id: sync_config.sync_run_id
60+
})
61+
write_failure += 1
62+
log_message_array << log_request_response("error", query, e.message)
63+
end
64+
end
65+
tracking_message(write_success, write_failure, log_message_array)
66+
rescue StandardError => e
67+
handle_exception(e, {
68+
context: "MYSQL:WRITE:EXCEPTION",
69+
type: "error",
70+
sync_id: sync_config.sync_id,
71+
sync_run_id: sync_config.sync_run_id
72+
})
73+
end
74+
75+
private
76+
77+
def create_connection(connection_config)
78+
Sequel.connect(
79+
adapter: "mysql2",
80+
host: connection_config[:host],
81+
port: connection_config[:port],
82+
user: connection_config[:username],
83+
password: connection_config[:password],
84+
database: connection_config[:database]
85+
)
86+
end
87+
88+
def create_streams(records)
89+
group_by_table(records).map do |_, r|
90+
Multiwoven::Integrations::Protocol::Stream.new(name: r[:tablename], action: StreamAction["fetch"], json_schema: convert_to_json_schema(r[:columns]))
91+
end
92+
end
93+
94+
def group_by_table(records)
95+
result = {}
96+
records.each_with_index do |entry, index|
97+
table_name = entry[:table_name]
98+
column_data = {
99+
column_name: entry[:column_name],
100+
data_type: entry[:data_type],
101+
is_nullable: entry[:is_nullable] == "YES"
102+
}
103+
result[index] ||= {}
104+
result[index][:tablename] = table_name
105+
result[index][:columns] = [column_data]
106+
end
107+
result.values.group_by { |entry| entry[:tablename] }.transform_values do |entries|
108+
{ tablename: entries.first[:tablename], columns: entries.flat_map { |entry| entry[:columns] } }
109+
end
110+
end
111+
end
112+
end
113+
end

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
{
2+
"data": {
3+
"name": "Mysql",
4+
"title": "MySQL",
5+
"connector_type": "destination",
6+
"category": "Database",
7+
"documentation_url": "https://docs.multiwoven.com/guides/destinations/mysql",
8+
"github_issue_label": "destination-mysql",
9+
"icon": "icon.svg",
10+
"license": "MIT",
11+
"release_stage": "alpha",
12+
"support_level": "community",
13+
"tags": ["language:ruby", "multiwoven"]
14+
}
15+
}

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

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
{
2+
"documentation_url": "https://docs.multiwoven.com/guides/destinations/mysql",
3+
"stream_type": "dynamic",
4+
"connector_query_type": "raw_sql",
5+
"connection_specification": {
6+
"$schema": "http://json-schema.org/draft-07/schema#",
7+
"title": "MySQL",
8+
"type": "object",
9+
"required": ["host", "port", "username", "password", "database"],
10+
"properties": {
11+
"host": {
12+
"description": "The hostname or IP address of the server where the MySQL database is hosted.",
13+
"examples": ["localhost"],
14+
"type": "string",
15+
"title": "Host",
16+
"order": 0
17+
},
18+
"port": {
19+
"description": "The port number on which the MySQL server is listening for connections.",
20+
"examples": ["3306"],
21+
"type": "string",
22+
"title": "Port",
23+
"order": 1
24+
},
25+
"username": {
26+
"description": "The username used to authenticate and connect to the MySQL database.",
27+
"examples": ["root"],
28+
"type": "string",
29+
"title": "Username",
30+
"order": 2
31+
},
32+
"password": {
33+
"description": "The password corresponding to the username used for authentication.",
34+
"type": "string",
35+
"multiwoven_secret": true,
36+
"title": "Password",
37+
"order": 3
38+
},
39+
"database": {
40+
"description": "The name of the specific database within the MySQL server to connect to.",
41+
"examples": ["mydatabase"],
42+
"type": "string",
43+
"title": "Database",
44+
"order": 4
45+
}
46+
}
47+
}
48+
}

‎integrations/lib/multiwoven/integrations/destination/mysql/icon.svg‎

Lines changed: 6 additions & 0 deletions
Loading

‎integrations/lib/multiwoven/integrations/rollout.rb‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
module Multiwoven
44
module Integrations
5-
VERSION = "0.35.0"
5+
VERSION = "0.35.1"
66

77
ENABLED_SOURCES = %w[
88
Snowflake
@@ -15,6 +15,7 @@ module Integrations
1515
Clickhouse
1616
AmazonS3
1717
MariaDB
18+
Mysql
1819
Oracle
1920
DatabricksModel
2021
AwsSagemakerModel
@@ -52,6 +53,7 @@ module Integrations
5253
Http
5354
Iterable
5455
MariaDB
56+
Mysql
5557
DatabricksLakehouse
5658
Oracle
5759
MicrosoftExcel

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

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
# frozen_string_literal: true
2+
3+
module Multiwoven::Integrations::Source
4+
module Mysql
5+
include Multiwoven::Integrations::Core
6+
class Client < SourceConnector
7+
def check_connection(connection_config)
8+
connection_config = connection_config.with_indifferent_access
9+
create_connection(connection_config)
10+
ConnectionStatus.new(status: ConnectionStatusType["succeeded"]).to_multiwoven_message
11+
rescue StandardError => e
12+
ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.message).to_multiwoven_message
13+
end
14+
15+
def discover(connection_config)
16+
connection_config = connection_config.with_indifferent_access
17+
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;"
18+
db = create_connection(connection_config)
19+
results = query_execution(db, query)
20+
catalog = Catalog.new(streams: create_streams(results))
21+
catalog.to_multiwoven_message
22+
rescue StandardError => e
23+
handle_exception(e, {
24+
context: "MYSQL:DISCOVER:EXCEPTION",
25+
type: "error"
26+
})
27+
end
28+
29+
def read(sync_config)
30+
connection_config = sync_config.source.connection_specification.with_indifferent_access
31+
query = sync_config.model.query
32+
query = batched_query(query, sync_config.limit, sync_config.offset) unless sync_config.limit.nil? && sync_config.offset.nil?
33+
db = create_connection(connection_config)
34+
query(db, query)
35+
rescue StandardError => e
36+
handle_exception(e, {
37+
context: "MYSQL:READ:EXCEPTION",
38+
type: "error",
39+
sync_id: sync_config.sync_id,
40+
sync_run_id: sync_config.sync_run_id
41+
})
42+
end
43+
44+
private
45+
46+
def create_connection(connection_config)
47+
Sequel.connect(
48+
adapter: "mysql2",
49+
host: connection_config[:host],
50+
port: connection_config[:port],
51+
user: connection_config[:username],
52+
password: connection_config[:password],
53+
database: connection_config[:database]
54+
)
55+
end
56+
57+
def query_execution(db, query)
58+
db.fetch(query).all
59+
end
60+
61+
def create_streams(records)
62+
group_by_table(records).map do |_, r|
63+
Multiwoven::Integrations::Protocol::Stream.new(name: r[:tablename], action: StreamAction["fetch"], json_schema: convert_to_json_schema(r[:columns]))
64+
end
65+
end
66+
67+
def query(db, query)
68+
records = []
69+
query_execution(db, query).map do |row|
70+
records << RecordMessage.new(data: row, emitted_at: Time.now.to_i).to_multiwoven_message
71+
end
72+
records
73+
end
74+
75+
def group_by_table(records)
76+
result = {}
77+
records.each_with_index do |entry, index|
78+
table_name = entry[:table_name]
79+
column_data = {
80+
column_name: entry[:column_name],
81+
data_type: entry[:data_type],
82+
is_nullable: entry[:is_nullable] == "YES"
83+
}
84+
result[index] ||= {}
85+
result[index][:tablename] = table_name
86+
result[index][:columns] = [column_data]
87+
end
88+
result
89+
end
90+
end
91+
end
92+
end

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
{
2+
"data": {
3+
"name": "Mysql",
4+
"title": "MySQL",
5+
"connector_type": "source",
6+
"category": "Data Warehouse",
7+
"sub_category": "Relational Database",
8+
"documentation_url": "https://docs.multiwoven.com/guides/sources/mysql",
9+
"github_issue_label": "source-mysql",
10+
"icon": "icon.svg",
11+
"license": "MIT",
12+
"release_stage": "alpha",
13+
"support_level": "community",
14+
"tags": ["language:ruby", "multiwoven"]
15+
}
16+
}

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

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
{
2+
"documentation_url": "https://docs.multiwoven.com/guides/sources/mysql",
3+
"stream_type": "dynamic",
4+
"connector_query_type": "raw_sql",
5+
"connection_specification": {
6+
"$schema": "http://json-schema.org/draft-07/schema#",
7+
"title": "MySQL",
8+
"type": "object",
9+
"required": ["host", "port", "username", "password", "database"],
10+
"properties": {
11+
"host": {
12+
"description": "The hostname or IP address of the server where the MySQL database is hosted.",
13+
"examples": ["localhost"],
14+
"type": "string",
15+
"title": "Host",
16+
"order": 0
17+
},
18+
"port": {
19+
"description": "The port number on which the MySQL server is listening for connections.",
20+
"examples": ["3306"],
21+
"type": "string",
22+
"title": "Port",
23+
"order": 1
24+
},
25+
"username": {
26+
"description": "The username used to authenticate and connect to the MySQL database.",
27+
"examples": ["root"],
28+
"type": "string",
29+
"title": "Username",
30+
"order": 2
31+
},
32+
"password": {
33+
"description": "The password corresponding to the username used for authentication.",
34+
"type": "string",
35+
"multiwoven_secret": true,
36+
"title": "Password",
37+
"order": 3
38+
},
39+
"database": {
40+
"description": "The name of the specific database within the MySQL server to connect to.",
41+
"examples": ["mydatabase"],
42+
"type": "string",
43+
"title": "Database",
44+
"order": 4
45+
}
46+
}
47+
}
48+
}

‎integrations/lib/multiwoven/integrations/source/mysql/icon.svg‎

Lines changed: 6 additions & 0 deletions
Loading

0 commit comments

Comments
 (0)