data_source.rb 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. require "digest/md5"
  2. module Blazer
  3. class DataSource
  4. attr_reader :id, :settings, :connection_model
  5. def initialize(id, settings)
  6. @id = id
  7. @settings = settings
  8. @connection_model =
  9. Class.new(Blazer::Connection) do
  10. def self.name
  11. "Blazer::Connection::#{object_id}"
  12. end
  13. establish_connection(settings["url"]) if settings["url"]
  14. end
  15. end
  16. def name
  17. settings["name"] || @id
  18. end
  19. def linked_columns
  20. settings["linked_columns"] || {}
  21. end
  22. def smart_columns
  23. settings["smart_columns"] || {}
  24. end
  25. def smart_variables
  26. settings["smart_variables"] || {}
  27. end
  28. def timeout
  29. settings["timeout"]
  30. end
  31. def cache
  32. settings["cache"]
  33. end
  34. def use_transaction?
  35. settings.key?("use_transaction") ? settings["use_transaction"] : true
  36. end
  37. def run_statement(statement, options = {})
  38. rows = nil
  39. error = nil
  40. cached_at = nil
  41. cache_key = self.cache_key(statement) if cache
  42. if cache && !options[:refresh_cache]
  43. value = Blazer.cache.read(cache_key)
  44. rows, cached_at = Marshal.load(value) if value
  45. end
  46. unless rows
  47. rows = []
  48. comment = "blazer"
  49. if options[:user].respond_to?(:id)
  50. comment << ",user_id:#{options[:user].id}"
  51. end
  52. if options[:user].respond_to?(Blazer.user_name)
  53. # only include letters, numbers, and spaces to prevent injection
  54. comment << ",user_name:#{options[:user].send(Blazer.user_name).to_s.gsub(/[^a-zA-Z0-9 ]/, "")}"
  55. end
  56. if options[:query].respond_to?(:id)
  57. comment << ",query_id:#{options[:query].id}"
  58. end
  59. in_transaction do
  60. begin
  61. connection_model.connection.execute("SET statement_timeout = #{timeout.to_i * 1000}") if timeout && (postgresql? || redshift?)
  62. result = connection_model.connection.select_all("#{statement} /*#{comment}*/")
  63. result.each do |untyped_row|
  64. row = {}
  65. untyped_row.each do |k, v|
  66. row[k] = result.column_types.empty? ? v : result.column_types[k].send(:type_cast, v)
  67. end
  68. rows << row
  69. end
  70. rescue ActiveRecord::StatementInvalid => e
  71. error = e.message.sub(/.+ERROR: /, "")
  72. error = Blazer::TIMEOUT_MESSAGE if error.include?("canceling statement due to statement timeout") || error.include?("cancelled on user's request") || error.include?("system requested abort query")
  73. end
  74. end
  75. Blazer.cache.write(cache_key, Marshal.dump([rows, Time.now]), expires_in: cache.to_f * 60) if !error && cache
  76. end
  77. [rows, error, cached_at]
  78. end
  79. def clear_cache(statement)
  80. Blazer.cache.delete(cache_key(statement))
  81. end
  82. def cache_key(statement)
  83. ["blazer", "v2", id, Digest::MD5.hexdigest(statement)].join("/")
  84. end
  85. def schemas
  86. default_schema = (postgresql? || redshift?) ? "public" : connection_model.connection_config[:database]
  87. settings["schemas"] || [connection_model.connection_config[:schema] || default_schema]
  88. end
  89. def tables
  90. rows, error, cached_at = run_statement(connection_model.send(:sanitize_sql_array, ["SELECT table_name, column_name, ordinal_position, data_type FROM information_schema.columns WHERE table_schema IN (?)", schemas]))
  91. Hash[rows.group_by { |r| r["table_name"] }.map { |t, f| [t, f.sort_by { |f| f["ordinal_position"] }.map { |f| f.slice("column_name", "data_type") }] }.sort_by { |t, _f| t }]
  92. end
  93. def postgresql?
  94. connection_model.connection.adapter_name == "PostgreSQL"
  95. end
  96. def redshift?
  97. connection_model.connection.adapter_name == "Redshift"
  98. end
  99. protected
  100. def in_transaction
  101. if use_transaction?
  102. connection_model.transaction do
  103. begin
  104. yield
  105. ensure
  106. raise ActiveRecord::Rollback
  107. end
  108. end
  109. else
  110. yield
  111. end
  112. end
  113. end
  114. end