module Neo4j module Server Neo4j::Session.register_db(:server_db) do |endpoint_url, url_opts| Neo4j::Server::CypherSession.open(endpoint_url, url_opts) end class CypherSession < Neo4j::Session include Resource include Neo4j::Core::CypherTranslator alias_method :super_query, :query attr_reader :connection def initialize(data_url, connection) @connection = connection Neo4j::Session.register(self) initialize_resource(data_url) Neo4j::Session._notify_listeners(:session_available, self) end # @param [Hash] params could be empty or contain basic authentication user and password # @return [Faraday] # @see https://github.com/lostisland/faraday def self.create_connection(params, url = nil) init_params = params[:initialize] && params.delete(:initialize) conn = Faraday.new(url, init_params) do |b| b.request :basic_auth, params[:basic_auth][:username], params[:basic_auth][:password] if params[:basic_auth] b.request :multi_json # b.response :logger b.response :multi_json, symbolize_keys: true, content_type: 'application/json' # b.use Faraday::Response::RaiseError b.use Faraday::Adapter::NetHttpPersistent # b.adapter Faraday.default_adapter end conn.headers = {'Content-Type' => 'application/json', 'User-Agent' => ::Neo4j::Session.user_agent_string} conn end # Opens a session to the database # @see Neo4j::Session#open # # @param [String] endpoint_url - the url to the neo4j server, defaults to 'http://localhost:7474' # @param [Hash] params faraday params, see #create_connection or an already created faraday connection def self.open(endpoint_url = nil, params = {}) extract_basic_auth(endpoint_url, params) url = endpoint_url || 'http://localhost:7474' connection = params[:connection] || create_connection(params, url) response = connection.get(url) fail "Server not available on #{url} (response code #{response.status})" unless response.status == 200 establish_session(response.body, connection) end def self.establish_session(root_data, connection) data_url = root_data[:data] data_url << '/' unless data_url.nil? || data_url.end_with?('/') CypherSession.new(data_url, connection) end def self.extract_basic_auth(url, params) return unless url && URI(url).userinfo params[:basic_auth] = {username: URI(url).user, password: URI(url).password} end private_class_method :extract_basic_auth def db_type :server_db end def to_s "#{self.class} url: '#{@resource_url}'" end def inspect "#{self} version: '#{version}'" end def version resource_data ? resource_data[:neo4j_version] : '' end def initialize_resource(data_url) response = @connection.get(data_url) expect_response_code!(response, 200) data_resource = response.body fail "No data_resource for #{response.body}" unless data_resource # store the resource data init_resource_data(data_resource, data_url) end def close super Neo4j::Transaction.unregister_current end def begin_tx Neo4j::Transaction.current ? Neo4j::Transaction.current.push_nested! : wrap_resource(@connection) Neo4j::Transaction.current end def create_node(props = nil, labels = []) id = _query_or_fail(cypher_string(labels, props), true, cypher_prop_list!(props)) value = props.nil? ? id : {id: id, metadata: {labels: labels}, data: props} CypherNode.new(self, value) end def load_node(neo_id) query.unwrapped.match(:n).where(n: {neo_id: neo_id}).pluck(:n).first end def load_relationship(neo_id) query.unwrapped.optional_match('(n)-[r]-()').where(r: {neo_id: neo_id}).pluck(:r).first rescue Neo4j::Session::CypherError => cypher_error if cypher_error.message.match(/not found$/) nil else raise cypher_error end end def create_label(name) CypherLabel.new(self, name) end def uniqueness_constraints(label) schema_properties("/db/data/schema/constraint/#{label}/uniqueness") end def indexes(label) schema_properties("/db/data/schema/index/#{label}") end def schema_properties(query_string) response = @connection.get(query_string) expect_response_code!(response, 200) {property_keys: response.body.map! { |row| row[:property_keys].map(&:to_sym) }} end def find_all_nodes(label_name) search_result_to_enumerable_first_column(_query_or_fail("MATCH (n:`#{label_name}`) RETURN ID(n)")) end def find_nodes(label_name, key, value) value = "'#{value}'" if value.is_a? String response = _query_or_fail("MATCH (n:`#{label_name}`) WHERE n.#{key} = #{value} RETURN ID(n)") search_result_to_enumerable_first_column(response) end def query(*args) if [[String], [String, Hash]].include?(args.map(&:class)) query = args[0] params = args[1] response = _query(query, params) response.raise_error if response.error? response.to_node_enumeration(query) else options = args[0] || {} Neo4j::Core::Query.new(options.merge(session: self)) end end def _query_data(query) r = _query_or_fail(query, true) Neo4j::Transaction.current ? r : r[:data] end DEFAULT_RETRY_COUNT = ENV['NEO4J_RETRY_COUNT'].nil? ? 10 : ENV['NEO4J_RETRY_COUNT'].to_i def _query_or_fail(query, single_row = false, params = {}, retry_count = DEFAULT_RETRY_COUNT) query, params = query_and_params(query, params) response = _query(query, params) if response.error? _retry_or_raise(query, params, single_row, retry_count, response) else single_row ? response.first_data : response end end def query_and_params(query_or_query_string, params) if query_or_query_string.is_a?(::Neo4j::Core::Query) cypher = query_or_query_string.to_cypher [cypher, query_or_query_string.send(:merge_params).merge(params)] else [query_or_query_string, params] end end def _retry_or_raise(query, params, single_row, retry_count, response) response.raise_error unless response.retryable_error? retry_count > 0 ? _query_or_fail(query, single_row, params, retry_count - 1) : response.raise_error end def _query_entity_data(query, id = nil, params = {}) _query(query, params).tap do |response| response.raise_error if response.error? end.entity_data(id) end def _query(query, params = {}, options = {}) query, params = query_and_params(query, params) curr_tx = Neo4j::Transaction.current ActiveSupport::Notifications.instrument('neo4j.cypher_query', context: options[:context] || 'CYPHER', cypher: query, params: params) do if curr_tx curr_tx._query(query, params) else url = resource_url(:cypher) query = params.nil? ? {'query' => query} : {'query' => query, 'params' => params} response = @connection.post(url, query) CypherResponse.create_with_no_tx(response) end end end def search_result_to_enumerable_first_column(response) return [] unless response.data Enumerator.new do |yielder| response.data.each do |data| if Neo4j::Transaction.current data[:row].each do |id| yielder << CypherNode.new(self, id).wrapper end else yielder << CypherNode.new(self, data[0]).wrapper end end end end EMPTY = '' def self.log_with clear, yellow, cyan = %W(\e[0m \e[33m \e[36m) ActiveSupport::Notifications.subscribe('neo4j.cypher_query') do |_, start, finish, _id, payload| ms = (finish - start) * 1000 params_string = (payload[:params] && payload[:params].size > 0 ? "| #{payload[:params].inspect}" : EMPTY) yield(" #{cyan}#{payload[:context]}#{clear} #{yellow}#{ms.round}ms#{clear} #{payload[:cypher]} #{params_string}") end end end end end