Merge pull request #22 from typesense/upsert-and-updates

Add support for upserts and update

Author: jasonbosco

README.md

@@ -33,7 +33,7 @@ Tests are also a good place to know how the the library works internally: [spec]
 
 | Typesense Server | typesense-ruby |
 |------------------|----------------|
-| \>= v0.15.1 | \>= v0.8.0 |
+| \>= v0.16.0 | \>= v0.8.0 |
 | \>= v0.15.0 | \>= v0.7.0 |
 | \>= v0.12.1 | \>= v0.5.0 |
 | \>= v0.12.0 | \>= v0.4.0 |

examples/client_initialization.rb

@@ -38,26 +38,26 @@
       host: 'localhost',
       port: 8108,
       protocol: 'http'
-    },
-    # Uncomment if starting a 3-node cluster, using Option 2 under Setup instructions above
-    {
-      host: 'localhost',
-      port: 7108,
-      protocol: 'http'
-    },
-    {
-      host: 'localhost',
-      port: 9108,
-      protocol: 'http'
     }
+    # Uncomment if starting a 3-node cluster, using Option 2 under Setup instructions above
+    # {
+    #   host: 'localhost',
+    #   port: 7108,
+    #   protocol: 'http'
+    # },
+    # {
+    #   host: 'localhost',
+    #   port: 9108,
+    #   protocol: 'http'
+    # }
   ],
   # If this optional key is specified, requests are always sent to this node first if it is healthy
   #   before falling back on the nodes mentioned in the `nodes` key. This is useful when running a distributed set of search clusters.
-  'nearest_node': {
-    'host': 'localhost',
-    'port': '8108',
-    'protocol': 'http'
-  },
+  # 'nearest_node': {
+  #   'host': 'localhost',
+  #   'port': '8108',
+  #   'protocol': 'http'
+  # },
   api_key: 'xyz',
   num_retries: 10,
   healthcheck_interval_seconds: 1,

examples/collections_and_documents.rb

@@ -167,6 +167,10 @@
 #   "num_employees" => 5215
 # }
 
+# You can also upsert a document, which will update the document if it already exists or create a new one if it doesn't exist
+document = @typesense.collections['companies'].documents.upsert(document)
+ap document
+
 ##
 # Retrieve a document
 sleep 0.5 # Give Typesense cluster a few hundred ms to create the document on all nodes, before reading it right after (eventually consistent)
@@ -180,6 +184,25 @@
 #   "num_employees" => 5215
 # }
 
+##
+# Update a document. Unlike upsert, update will error out if the doc doesn't already exist.
+document = @typesense.collections['companies'].documents['124'].update(
+  'id' => 1,
+  'num_employees' => 5500
+)
+ap document
+
+# {
+#   "id"            => "124",
+#   "num_employees" => 5500
+# }
+
+# This should error out, since document 145 doesn't exist
+# document = @typesense.collections['companies'].documents['145'].update(
+#   'num_employees' => 5500
+# )
+# ap document
+
 ##
 # Delete a document
 #   Deleting a document, returns the document after deletion
@@ -208,11 +231,33 @@
     'country' => 'France'
   }
 ]
-ap @typesense.collections['companies'].documents.create_many(documents)
+ap @typesense.collections['companies'].documents.import(documents)
 
-## If you already have documents in JSONL format, you can also use #import instead, to avoid the JSON parsing overhead:
+## If you already have documents in JSONL format, you can also pass it directly to #import, to avoid the JSON parsing overhead:
 # @typesense.collections['companies'].documents.import(documents_in_jsonl_format)
 
+## You can bulk upsert documents, by adding an upsert action option to #import
+documents << {
+  'id' => '126',
+  'company_name' => 'Stark Industries 2',
+  'num_employees' => 200,
+  'country' => 'USA'
+}
+ap @typesense.collections['companies'].documents.import(documents, action: :upsert)
+
+## You can bulk update documents, by adding an update action option to #import
+# `action: update` will throw an error if the document doesn't already exist
+# This document will error out, since id: 1200 doesn't exist
+documents << {
+  'id' => '1200',
+  'country' => 'USA'
+}
+documents << {
+  'id' => '126',
+  'num_employees' => 300
+}
+ap @typesense.collections['companies'].documents.import(documents, action: :update)
+
 ##
 # Export all documents in a collection in JSON Lines format
 #   We use JSON Lines format for performance reasons. You can choose to parse selected lines as needed, by splitting on \n.

lib/typesense/api_call.rb

@@ -7,6 +7,8 @@ module Typesense
   class ApiCall
     API_KEY_HEADER_NAME = 'X-TYPESENSE-API-KEY'
 
+    attr_reader :logger
+
     def initialize(configuration)
       @configuration = configuration
 
@@ -24,45 +26,40 @@ def initialize(configuration)
       @current_node_index = -1
     end
 
-    def post(endpoint, parameters = {})
-      headers, query, body = split_post_put_parameters(parameters)
-
+    def post(endpoint, body_parameters = {}, query_parameters = {})
       perform_request :post,
                       endpoint,
-                      headers,
-                      params: query,
-                      body: body
+                      query_parameters: query_parameters,
+                      body_parameters: body_parameters
     end
 
-    def put(endpoint, parameters = {})
-      headers, query, body = split_post_put_parameters(parameters)
+    def patch(endpoint, body_parameters = {}, query_parameters = {})
+      perform_request :patch,
+                      endpoint,
+                      query_parameters: query_parameters,
+                      body_parameters: body_parameters
+    end
 
+    def put(endpoint, body_parameters = {}, query_parameters = {})
       perform_request :put,
                       endpoint,
-                      headers,
-                      params: query,
-                      body: body
+                      query_parameters: query_parameters,
+                      body_parameters: body_parameters
     end
 
-    def get(endpoint, parameters = {})
-      headers, query = extract_headers_and_query_from(parameters)
-
+    def get(endpoint, query_parameters = {})
       perform_request :get,
                       endpoint,
-                      headers,
-                      params: query
+                      query_parameters: query_parameters
     end
 
-    def delete(endpoint, parameters = {})
-      headers, query = extract_headers_and_query_from(parameters)
-
+    def delete(endpoint, query_parameters = {})
       perform_request :delete,
                       endpoint,
-                      headers,
-                      params: query
+                      query_parameters: query_parameters
     end
 
-    def perform_request(method, endpoint, headers = {}, options = {})
+    def perform_request(method, endpoint, query_parameters: nil, body_parameters: nil, additional_headers: {})
       @configuration.validate!
       last_exception = nil
       @logger.debug "Performing #{method.to_s.upcase} request: #{endpoint}"
@@ -72,15 +69,23 @@ def perform_request(method, endpoint, headers = {}, options = {})
         @logger.debug "Attempting #{method.to_s.upcase} request Try ##{num_tries} to Node #{node[:index]}"
 
         begin
-          response = Typhoeus::Request.new(uri_for(endpoint, node),
-                                           {
-                                             method: method,
-                                             headers: default_headers.merge(headers),
-                                             timeout: @connection_timeout_seconds
-                                           }.merge(options)).run
+          request_options = {
+            method: method,
+            timeout: @connection_timeout_seconds,
+            headers: default_headers.merge(additional_headers)
+          }
+          request_options.merge!(params: query_parameters) unless query_parameters.nil?
+
+          unless body_parameters.nil?
+            body = body_parameters
+            body = Oj.dump(body_parameters) if request_options[:headers]['Content-Type'] == 'application/json'
+            request_options.merge!(body: body)
+          end
+
+          response = Typhoeus::Request.new(uri_for(endpoint, node), request_options).run
           set_node_healthcheck(node, is_healthy: true) if response.code >= 1 && response.code <= 499
 
-          @logger.debug "Request to Node #{node[:index]} was successfully made (at the network layer). Response Code was #{response.code}."
+          @logger.debug "Request #{method}:#{uri_for(endpoint, node)} to Node #{node[:index]} was successfully made (at the network layer). Response Code was #{response.code}."
 
           parsed_response = if response.headers && (response.headers['content-type'] || '').include?('application/json')
                               Oj.load(response.body)
@@ -101,7 +106,7 @@ def perform_request(method, endpoint, headers = {}, options = {})
           #   other languages that might not support the same construct.
           set_node_healthcheck(node, is_healthy: false)
           last_exception = e
-          @logger.warn "Request to Node #{node[:index]} failed due to \"#{e.class}: #{e.message}\""
+          @logger.warn "Request #{method}:#{uri_for(endpoint, node)} to Node #{node[:index]} failed due to \"#{e.class}: #{e.message}\""
           @logger.warn "Sleeping for #{@retry_interval_seconds}s and then retrying request..."
           sleep @retry_interval_seconds
         end
@@ -112,43 +117,6 @@ def perform_request(method, endpoint, headers = {}, options = {})
 
     private
 
-    def split_post_put_parameters(parameters)
-      if json_request?(parameters)
-        headers = { 'Content-Type' => 'application/json' }
-        query = parameters[:query]
-        body = Oj.dump(sanitize_parameters(parameters))
-      else
-        headers = {}
-        query = parameters[:query]
-        body = parameters[:body]
-      end
-      [headers, query, body]
-    end
-
-    def extract_headers_and_query_from(parameters)
-      if json_request?(parameters)
-        headers = { 'Content-Type' => 'application/json' }
-        query = sanitize_parameters(parameters)
-      else
-        headers = {}
-        query = parameters[:query]
-      end
-      [headers, query]
-    end
-
-    def json_request?(parameters)
-      parameters[:as_json].nil? ? true : parameters[:as_json]
-    end
-
-    def sanitize_parameters(parameters)
-      sanitized_parameters = parameters.dup
-      sanitized_parameters.delete(:as_json)
-      sanitized_parameters.delete(:body)
-      sanitized_parameters.delete(:query)
-
-      sanitized_parameters
-    end
-
     def uri_for(endpoint, node)
       "#{node[:protocol]}://#{node[:host]}:#{node[:port]}#{endpoint}"
     end
@@ -231,6 +199,7 @@ def custom_exception_klass_for(response)
 
     def default_headers
       {
+        'Content-Type' => 'application/json',
         API_KEY_HEADER_NAME.to_s => @api_key,
         'User-Agent' => 'Typesense Ruby Client'
       }

lib/typesense/document.rb

@@ -16,6 +16,10 @@ def delete
       @api_call.delete(endpoint_path)
     end
 
+    def update(partial_document)
+      @api_call.patch(endpoint_path, partial_document)
+    end
+
     private
 
     def endpoint_path

lib/typesense/documents.rb

@@ -16,17 +16,40 @@ def create(document)
       @api_call.post(endpoint_path, document)
     end
 
-    def create_many(documents)
-      documents_in_jsonl_format = documents.map { |document| Oj.dump(document) }.join("\n")
-      results_in_jsonl_format = import(documents_in_jsonl_format)
-      results_in_jsonl_format.split("\n").map { |r| Oj.load(r) }
+    def upsert(document)
+      @api_call.post(endpoint_path, document, action: :upsert)
     end
 
-    def import(documents_in_jsonl_format, query_parameters = {})
-      @api_call.post(endpoint_path('import'),
-                     as_json: false,
-                     query: query_parameters,
-                     body: documents_in_jsonl_format)
+    def update(document)
+      @api_call.post(endpoint_path, document, action: :update)
+    end
+
+    def create_many(documents, options = {})
+      @api_call.logger.warn('#create_many is deprecated and will be removed in a future version. Use #import instead, which now takes both an array of documents or a JSONL string of documents')
+      import(documents, options)
+    end
+
+    # @param [Array,String] documents An array of document hashes or a JSONL string of documents.
+    def import(documents, options = {})
+      documents_in_jsonl_format = if documents.is_a?(Array)
+                                    documents.map { |document| Oj.dump(document) }.join("\n")
+                                  else
+                                    documents
+                                  end
+
+      results_in_jsonl_format = @api_call.perform_request(
+        'post',
+        endpoint_path('import'),
+        query_parameters: options,
+        body_parameters: documents_in_jsonl_format,
+        additional_headers: { 'Content-Type' => 'text/plain' }
+      )
+
+      if documents.is_a?(Array)
+        results_in_jsonl_format.split("\n").map { |r| Oj.load(r) }
+      else
+        results_in_jsonl_format
+      end
     end
 
     def export

spec/typesense/document_spec.rb

@@ -57,6 +57,27 @@
     end
   end
 
+  describe '#update' do
+    it 'updates the specified document' do
+      partial_document = {
+        'id' => '124',
+        'num_employees' => 5200
+      }
+      stub_request(:patch, Typesense::ApiCall.new(typesense.configuration).send(:uri_for, '/collections/companies/documents/124', typesense.configuration.nodes[0]))
+        .with(
+          headers: {
+            'Content-Type' => 'application/json',
+            'X-Typesense-Api-Key' => typesense.configuration.api_key
+          }
+        )
+        .to_return(status: 200, body: JSON.dump(partial_document), headers: { 'Content-Type': 'application/json' })
+
+      result = document_124.update(partial_document)
+
+      expect(result).to eq(partial_document)
+    end
+  end
+
   describe '#delete' do
     it 'deletes the specified document' do
       stub_request(:delete, Typesense::ApiCall.new(typesense.configuration).send(:uri_for, '/collections/companies/documents/124', typesense.configuration.nodes[0]))