Mofonetprofiles 294 create dox graph db consumer batching

CHANGELOG.md

@@ -1,6 +1,9 @@
 Changelog
 =========
 
+## v1.2.0 09/15/2022
+* Adds a new batch client. See README for details.
+
 ## v1.1.0 07/26/2022
 * Adds ability for relationships to function with a primary key attr
 

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    neo4j-http (1.1.0)
+    neo4j-http (1.2.0)
       activesupport (>= 5.2)
       faraday (< 2)
       faraday-retry
@@ -11,7 +11,7 @@ PATH
 GEM
   remote: https://rubygems.org/
   specs:
-    activesupport (7.0.3.1)
+    activesupport (7.0.4)
       concurrent-ruby (~> 1.0, >= 1.0.2)
       i18n (>= 1.6, < 2)
       minitest (>= 5.1)
@@ -20,7 +20,7 @@ GEM
     coderay (1.1.3)
     concurrent-ruby (1.1.10)
     diff-lcs (1.5.0)
-    faraday (1.10.0)
+    faraday (1.10.2)
       faraday-em_http (~> 1.0)
       faraday-em_synchrony (~> 1.0)
       faraday-excon (~> 1.1)
@@ -48,7 +48,7 @@ GEM
     i18n (1.12.0)
       concurrent-ruby (~> 1.0)
     method_source (1.0.0)
-    minitest (5.16.2)
+    minitest (5.16.3)
     multipart-post (2.2.3)
     parallel (1.22.1)
     parser (3.1.1.0)
@@ -92,7 +92,7 @@ GEM
     standard (1.9.1)
       rubocop (= 1.26.1)
       rubocop-performance (= 1.13.3)
-    tzinfo (2.0.4)
+    tzinfo (2.0.5)
       concurrent-ruby (~> 1.0)
     unicode-display_width (2.1.0)
 

README.md

@@ -124,6 +124,29 @@ cypher_client = Neo4j::Http::CypherClient.new(config)
 node_client = Neo4j::Http::NodeClient.new(cypher_client)
 ```
 
+## Batch operations
+
+The `Neo4j::Http::Client.in_batch` will yield a batch client. It can be used like:
+
+```ruby
+Neo4j::Http::Client.in_batch do |tx|
+  [
+    tx.upsert_node(node),
+    tx.upsert_node(node2),
+    tx.upsert_relationship(relationship: relationship, from: from, to: to)
+  ]
+end
+```
+
+All of the commands need to chain off of the variable exposed by the block in order to
+prepare the operations for the batch. These are not immediately invoked like their
+single operation counterparts. The syntax and arguments are identical.
+
+The array of statements will be passed into a batch client that will
+prepare the statements and the parameters and issue a single
+request to the Neo4j HTTP API. Note that the size of the batch is
+determined by the caller's array length.
+
 ## Versioning
 
 This project follows [semantic versioning](https://semver.org).

lib/neo4j/http.rb

@@ -8,13 +8,17 @@
 
 require "neo4j/http/auth_token"
 require "neo4j/http/client"
+require "neo4j/http/batch_client"
 require "neo4j/http/configuration"
 require "neo4j/http/cypher_client"
+require "neo4j/http/batch_cypher_client"
 require "neo4j/http/object_wrapper"
 require "neo4j/http/node"
 require "neo4j/http/node_client"
+require "neo4j/http/batch_node_client"
 require "neo4j/http/relationship"
 require "neo4j/http/relationship_client"
+require "neo4j/http/batch_relationship_client"
 require "neo4j/http/results"
 
 require "neo4j/http/errors"
@@ -24,7 +28,7 @@ module Http
     extend self
 
     def config
-      @congiguration ||= Configuration.new
+      @configuration ||= Configuration.new
     end
 
     def configure

lib/neo4j/http/batch_client.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+module Neo4j
+  module Http
+    class BatchClient < Client
+      class << self
+        def default
+          cypher_client = Http::BatchCypherClient.new(Neo4j::Http.config)
+          node_client = Http::BatchNodeClient.new(cypher_client)
+          relationship_client = Http::BatchRelationshipClient.new(cypher_client)
+          @default ||= new(cypher_client, node_client, relationship_client)
+        end
+      end
+
+      attr_accessor :cypher_client, :node_client, :relationship_client
+
+      def initialize(cypher_client, node_client, relationship_client)
+        @cypher_client = cypher_client
+        @node_client = node_client
+        @relationship_client = relationship_client
+      end
+
+      delegate(*CYPHER_CLIENT_METHODS, to: :cypher_client)
+      delegate(*NODE_CLIENT_METHODS, to: :node_client)
+      delegate(*RELATIONSHIP_CLIENT_METHODS, to: :relationship_client)
+    end
+  end
+end

lib/neo4j/http/batch_cypher_client.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+require "forwardable"
+require "faraday"
+require "faraday/retry"
+require "faraday_middleware"
+
+module Neo4j
+  module Http
+    class BatchCypherClient < CypherClient
+      # Each statement should be Hash with statement and parameters keys e.g.
+      #   {
+      #     statement: "MATCH (n:User { name: $name }) RETURN n",
+      #     parameters: { name: "Ben" }
+      #   }
+      # https://neo4j.com/docs/http-api/current/actions/execute-multiple-statements/
+      def execute_cypher(statements = [])
+        statements = [statements] if statements.is_a?(Hash) # equivalent to Array.wrap
+
+        request_body = {
+          statements: statements.map do |statement|
+            {
+              statement: statement[:statement],
+              parameters: statement[:parameters].as_json
+            }
+          end
+        }
+
+        @connection = @injected_connection || connection("WRITE")
+        response = @connection.post(transaction_path, request_body)
+        results = check_errors!(statements, response)
+
+        results.map do |result|
+          Neo4j::Http::Results.parse(result || {})
+        end
+      end
+    end
+  end
+end

lib/neo4j/http/batch_node_client.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+module Neo4j
+  module Http
+    class BatchNodeClient < NodeClient
+      protected
+
+      def process_upsert_node(cypher:, node:)
+        {
+          statement: cypher,
+          parameters: {key_value: node.key_value, attributes: node.attributes}
+        }
+      end
+
+      def process_delete_node(cypher:, node:)
+        {
+          statement: cypher,
+          parameters: {key_value: node.key_value}
+        }
+      end
+    end
+  end
+end

lib/neo4j/http/batch_relationship_client.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module Neo4j
+  module Http
+    class BatchRelationshipClient < RelationshipClient
+      protected
+
+      def process_upsert_relationship(cypher:, from:, to:, relationship:)
+        {
+          statement: cypher,
+          parameters: {
+            from: from,
+            to: to,
+            relationship: relationship,
+            relationship_attributes: relationship.attributes
+          }
+        }
+      end
+
+      def process_delete_relationship(cypher:, from:, to:)
+        {
+          statement: cypher,
+          parameters: {
+            from: from,
+            to: to
+          }
+        }
+      end
+
+      def process_delete_relationship_on_primary_key(cypher:, relationship:)
+        {
+          statement: cypher,
+          parameters: {
+            relationship: relationship
+          }
+        }
+      end
+    end
+  end
+end

lib/neo4j/http/client.rb

@@ -17,6 +17,11 @@ def default
           relationship_client = Http::RelationshipClient.new(cypher_client)
           @default ||= new(cypher_client, node_client, relationship_client)
         end
+
+        def in_batch &block
+          batch_client = Neo4j::Http::BatchClient
+          batch_client.execute_cypher yield(batch_client)
+        end
       end
 
       attr_accessor :cypher_client, :node_client, :relationship_client

lib/neo4j/http/cypher_client.rb

@@ -47,7 +47,7 @@ def connection(access_mode)
       protected
 
       delegate :auth_token, :transaction_path, to: :@configuration
-      def check_errors!(cypher, response, parameters)
+      def check_errors!(cypher, response, parameters = {})
         raise Neo4j::Http::Errors::InvalidConnectionUrl, response.status if response.status == 404
         if response.body["errors"].any? { |error| error["message"][/Routing WRITE queries is not supported/] }
           raise Neo4j::Http::Errors::ReadOnlyError

lib/neo4j/http/node_client.rb

@@ -21,9 +21,7 @@ def upsert_node(node)
           return node
         CYPHER
 
-        results = @cypher_client.execute_cypher(cypher, key_value: node.key_value, attributes: node.attributes)
-
-        results.first&.fetch("node")
+        process_upsert_node(cypher: cypher, node: node)
       end
 
       def delete_node(node)
@@ -34,8 +32,7 @@ def delete_node(node)
           RETURN node
         CYPHER
 
-        results = @cypher_client.execute_cypher(cypher, key_value: node.key_value)
-        results.first&.fetch("node")
+        process_delete_node(cypher: cypher, node: node)
       end
 
       def find_node_by(label:, **attributes)
@@ -56,6 +53,16 @@ def find_nodes_by(label:, attributes:, limit: 100)
 
       protected
 
+      def process_upsert_node(cypher:, node:)
+        results = @cypher_client.execute_cypher(cypher, key_value: node.key_value, attributes: node.attributes)
+        results.first&.fetch("node")
+      end
+
+      def process_delete_node(cypher:, node:)
+        results = @cypher_client.execute_cypher(cypher, key_value: node.key_value)
+        results.first&.fetch("node")
+      end
+
       def build_selectors(attributes, node_name: :node)
         attributes.map do |key, value|
           if value.is_a?(Array)

lib/neo4j/http/relationship_client.rb

@@ -33,14 +33,7 @@ def upsert_relationship(relationship:, from:, to:, create_nodes: false)
           RETURN from, to, relationship
         CYPHER
 
-        results = @cypher_client.execute_cypher(
-          cypher,
-          from: from,
-          to: to,
-          relationship: relationship,
-          relationship_attributes: relationship.attributes
-        )
-        results&.first
+        process_upsert_relationship(cypher: cypher, from: from, to: to, relationship: relationship)
       end
 
       def find_relationships(from:, relationship:, to:)
@@ -84,8 +77,7 @@ def delete_relationship(relationship:, from:, to:)
           RETURN from, to
         CYPHER
 
-        results = @cypher_client.execute_cypher(cypher, from: from, to: to)
-        results&.first
+        process_delete_relationship(cypher: cypher, from: from, to: to)
       end
 
       def delete_relationship_on_primary_key(relationship:)
@@ -101,6 +93,28 @@ def delete_relationship_on_primary_key(relationship:)
           RETURN relationship
         CYPHER
 
+        process_delete_relationship_on_primary_key(cypher: cypher, relationship: relationship)
+      end
+
+      protected
+
+      def process_upsert_relationship(cypher:, from:, to:, relationship:)
+        results = @cypher_client.execute_cypher(
+          cypher,
+          from: from,
+          to: to,
+          relationship: relationship,
+          relationship_attributes: relationship.attributes
+        )
+        results&.first
+      end
+
+      def process_delete_relationship(cypher:, from:, to:)
+        results = @cypher_client.execute_cypher(cypher, from: from, to: to)
+        results&.first
+      end
+
+      def process_delete_relationship_on_primary_key(cypher:, relationship:)
         results = @cypher_client.execute_cypher(cypher, relationship: relationship)
         results&.first
       end

lib/neo4j/http/version.rb

@@ -1,5 +1,5 @@
 module Neo4j
   module Http
-    VERSION = "1.1.0"
+    VERSION = "1.2.0"
   end
 end