Class: Multiwoven::Integrations::Source::Snowflake::Client
- Inherits:
-
SourceConnector
- Object
- SourceConnector
- Multiwoven::Integrations::Source::Snowflake::Client
- Defined in:
- lib/multiwoven/integrations/source/snowflake/client.rb
Instance Method Summary collapse
- #check_connection(connection_config) ⇒ Object
- #discover(connection_config) ⇒ Object
- #read(sync_config) ⇒ Object
Instance Method Details
#check_connection(connection_config) ⇒ Object
7 8 9 10 11 12 |
# File 'lib/multiwoven/integrations/source/snowflake/client.rb', line 7 def check_connection(connection_config) create_connection(connection_config) ConnectionStatus.new(status: ConnectionStatusType["succeeded"]). rescue Sequel::DatabaseConnectionError => e ConnectionStatus.new(status: ConnectionStatusType["failed"], message: e.). end |
#discover(connection_config) ⇒ Object
14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
# File 'lib/multiwoven/integrations/source/snowflake/client.rb', line 14 def discover(connection_config) query = "SELECT table_name, column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = \'#{connection_config[:schema]}\' AND table_catalog = \'#{connection_config[:database]}\' ORDER BY table_name, ordinal_position;" db = create_connection(connection_config) records = [] db.fetch(query.gsub("\n", "")) do |row| records << row end catalog = Catalog.new(streams: create_streams(records)) catalog. rescue StandardError => e handle_exception( "SNOWFLAKE:DISCOVER:EXCEPTION", "error", e ) end |
#read(sync_config) ⇒ Object
36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
# File 'lib/multiwoven/integrations/source/snowflake/client.rb', line 36 def read(sync_config) connection_config = sync_config.source.connection_specification query = sync_config.model.query db = create_connection(connection_config) records = [] db.fetch(query) do |row| records << RecordMessage.new(data: row, emitted_at: Time.now.to_i). end records rescue StandardError => e handle_exception( "SNOWFLAKE:READ:EXCEPTION", "error", e ) end |