mirror of
https://github.com/ToolJet/ToolJet
synced 2026-05-21 16:08:35 +00:00
113 lines
2.9 KiB
Ruby
113 lines
2.9 KiB
Ruby
class DynamodbQueryService
|
|
attr_accessor :data_query, :data_source, :options, :source_options, :current_user
|
|
|
|
def initialize(data_query, data_source, options, source_options, current_user)
|
|
@data_query = data_query
|
|
@data_source = data_source
|
|
@options = options
|
|
@source_options = source_options
|
|
@current_user = current_user
|
|
end
|
|
|
|
def self.connection options
|
|
|
|
region = options.dig('region', 'value')
|
|
access_key = options.dig('access_key', 'value')
|
|
secret_key = options.dig('secret_key', 'value')
|
|
|
|
credentials = Aws::Credentials.new(access_key, secret_key)
|
|
dynamodb = Aws::DynamoDB::Client.new(region: region, credentials: credentials)
|
|
|
|
dynamodb.list_tables
|
|
end
|
|
|
|
def process
|
|
error = nil
|
|
data = []
|
|
operation = options["operation"]
|
|
|
|
begin
|
|
|
|
connection = get_connection
|
|
data = send("exec_#{operation}", connection, options)
|
|
|
|
rescue StandardError => e
|
|
puts e
|
|
error = e.message
|
|
end
|
|
|
|
{ status: error ? 'failed' : 'success', data: data, error: { message: error } }
|
|
end
|
|
|
|
private
|
|
def get_connection
|
|
if $connections.include? data_source.id
|
|
connection = $connections[data_source.id][:connection]
|
|
else
|
|
region = source_options['region']
|
|
access_key = source_options['access_key']
|
|
secret_key = source_options['secret_key']
|
|
|
|
credentials = Aws::Credentials.new(access_key, secret_key)
|
|
connection = Aws::DynamoDB::Client.new(region: region, credentials: credentials)
|
|
|
|
$connections[data_source.id] = { connection: connection }
|
|
end
|
|
|
|
connection
|
|
end
|
|
|
|
def exec_list_tables(connection, options)
|
|
tables = connection.list_tables
|
|
tables.to_h
|
|
end
|
|
|
|
def exec_get_item(connection, options)
|
|
table = options["table"]
|
|
key = JSON.parse(options["key"])
|
|
|
|
item = {
|
|
table_name: table,
|
|
key: key
|
|
}
|
|
|
|
connection.get_item(item).to_h
|
|
end
|
|
|
|
def exec_delete_item(connection, options)
|
|
table = options["table"]
|
|
key = JSON.parse(options["key"])
|
|
|
|
item = {
|
|
table_name: table,
|
|
key: key
|
|
}
|
|
data = connection.delete_item(item).to_h
|
|
end
|
|
|
|
def exec_query_table(connection, options)
|
|
query_condition = JSON.parse(options["query_condition"])
|
|
query_condition = query_condition.transform_keys(&:to_sym) # DynamoDB SDK requires the keys as syms
|
|
|
|
result = connection.query(query_condition)
|
|
data = []
|
|
result.items.each do |item|
|
|
data << item
|
|
end
|
|
|
|
data
|
|
end
|
|
|
|
def exec_scan_table(connection, options)
|
|
scan_condition = JSON.parse(options["scan_condition"])
|
|
scan_condition = scan_condition.transform_keys(&:to_sym) # DynamoDB SDK requires the keys as syms
|
|
|
|
result = connection.scan(scan_condition)
|
|
data = []
|
|
result.items.each do |item|
|
|
data << item
|
|
end
|
|
|
|
data
|
|
end
|
|
end
|