Sha256: 1e4c02f4bbbb88e47f6ba461b0a4955fe04d0c91755bf8d622c4c995ded64e18
Contents?: true
Size: 1.45 KB
Versions: 20
Compression:
Stored size: 1.45 KB
Contents
module Remi
module DataSource
class Postgres
include DataSource
def initialize(fields: {}, credentials:, query:, logger: Remi::Settings.logger)
@fields = fields
@credentials = credentials
@query = query
@logger = logger
end
attr_accessor :fields
def extract
@logger.info "Executing query #{@query}"
@raw_result = pg_conn.exec @query
end
def raw_result
@raw_result ||= extract
end
def pg_conn
@pg_conn ||= PG.connect(
host: @credentials[:host] || 'localhost',
port: @credentials[:port] || 5432,
dbname: @credentials[:dbname],
user: @credentials[:user] || `whoami`.chomp,
password: @credentials[:password],
sslmode: @credentials[:sslmode] || 'require'
)
end
def to_dataframe
# Performance for larger sets could be improved by using bulk query (via COPY)
@logger.info "Converting query to a dataframe"
hash_array = {}
raw_result.each do |row|
row.each do |field, value|
(hash_array[field_symbolizer.call(field)] ||= []) << value
end
end
# After converting to DF, clear the PG results to save memory.
raw_result.clear
Daru::DataFrame.new hash_array, order: hash_array.keys
end
def df
@dataframe ||= to_dataframe
end
end
end
end
Version data entries
20 entries across 20 versions & 1 rubygems