Feat/add-http-destination-connector

integrations/lib/multiwoven/integrations.rb

@@ -61,6 +61,7 @@
 require_relative "integrations/destination/salesforce_consumer_goods_cloud/client"
 require_relative "integrations/destination/sftp/client"
 require_relative "integrations/destination/postgresql/client"
+require_relative "integrations/destination/http/client"
 
 module Multiwoven
   module Integrations

integrations/lib/multiwoven/integrations/destination/http/client.rb

@@ -0,0 +1,95 @@
+# frozen_string_literal: true
+
+module Multiwoven
+  module Integrations
+    module Destination
+      module Http
+        include Multiwoven::Integrations::Core
+        class Client < DestinationConnector
+          MAX_CHUNK_SIZE = 10
+          def check_connection(connection_config)
+            connection_config = connection_config.with_indifferent_access
+            destination_url = connection_config[:destination_url]
+            headers = connection_config[:headers]
+            request = Multiwoven::Integrations::Core::HttpClient.request(
+              destination_url,
+              HTTP_POST,
+              payload: {},
+              headers: headers
+            )
+            if success?(request)
+              success_status
+            else
+              failure_status(nil)
+            end
+          rescue StandardError => e
+            handle_exception("HTTP:CHECK_CONNECTION:EXCEPTION", "error", e)
+            failure_status(e)
+          end
+
+          def discover(_connection_config = nil)
+            catalog_json = read_json(CATALOG_SPEC_PATH)
+            catalog = build_catalog(catalog_json)
+            catalog.to_multiwoven_message
+          rescue StandardError => e
+            handle_exception(
+              "HTTP:DISCOVER:EXCEPTION",
+              "error",
+              e
+            )
+          end
+
+          def write(sync_config, records, _action = "create")
+            connection_config = sync_config.destination.connection_specification.with_indifferent_access
+            url = connection_config[:destination_url]
+            headers = connection_config[:headers]
+            write_success = 0
+            write_failure = 0
+            records.each_slice(MAX_CHUNK_SIZE) do |chunk|
+              payload = create_payload(chunk)
+              response = Multiwoven::Integrations::Core::HttpClient.request(
+                url,
+                sync_config.stream.request_method,
+                payload: payload,
+                headers: headers
+              )
+              if success?(response)
+                write_success += chunk.size
+              else
+                write_failure += chunk.size
+              end
+            rescue StandardError => e
+              handle_exception("HTTP:RECORD:WRITE:EXCEPTION", "error", e)
+              write_failure += chunk.size
+            end
+
+            tracker = Multiwoven::Integrations::Protocol::TrackingMessage.new(
+              success: write_success,
+              failed: write_failure
+            )
+            tracker.to_multiwoven_message
+          rescue StandardError => e
+            handle_exception("HTTP:WRITE:EXCEPTION", "error", e)
+          end
+
+          private
+
+          def create_payload(records)
+            {
+              "records" => records.map do |record|
+                {
+                  "fields" => record
+                }
+              end
+            }
+          end
+
+          def extract_body(response)
+            response_body = response.body
+            JSON.parse(response_body) if response_body
+          end
+        end
+      end
+    end
+  end
+end

integrations/lib/multiwoven/integrations/destination/http/config/catalog.json

@@ -0,0 +1,16 @@
+{
+  "request_rate_limit": 600,
+  "request_rate_limit_unit": "minute",
+  "request_rate_concurrency": 10,
+  "schema_mode": "schemaless",
+  "streams": [
+    {
+      "name": "http",
+      "batch_support": false,
+      "action": "create",
+      "method": "POST",
+      "json_schema": {},
+      "supported_sync_modes": ["incremental"]
+    }
+  ]
+}

integrations/lib/multiwoven/integrations/destination/http/config/meta.json

@@ -0,0 +1,15 @@
+{
+  "data": {
+    "name": "Http",
+    "title": "http",
+    "connector_type": "destination",
+    "category": "Http",
+    "documentation_url": "https://docs.multiwoven.com/destinations/http",
+    "github_issue_label": "destination-http",
+    "icon": "icon.svg",
+    "license": "MIT",
+    "release_stage": "alpha",
+    "support_level": "community",
+    "tags": ["language:ruby", "multiwoven"]
+  }
+}

integrations/lib/multiwoven/integrations/destination/http/config/spec.json

@@ -0,0 +1,24 @@
+{
+  "documentation_url": "https://docs.multiwoven.com/integrations/destination/http",
+  "stream_type": "static",
+  "connection_specification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "Http",
+    "type": "object",
+    "required": ["destination_url"],
+    "properties": {
+      "destination_url": {
+        "type": "string",
+        "title": "Destination url",
+        "order": 0
+      },
+      "headers": {
+        "title": "Http Headers",
+        "order": 1,
+        "additionalProperties": {
+          "type": "string"
+        }
+      }
+    }
+  }
+}

integrations/lib/multiwoven/integrations/rollout.rb

@@ -2,7 +2,7 @@
 
 module Multiwoven
   module Integrations
-    VERSION = "0.1.64"
+    VERSION = "0.1.65"
 
     ENABLED_SOURCES = %w[
       Snowflake
@@ -25,6 +25,7 @@ module Integrations
       SalesforceConsumerGoodsCloud
       Sftp
       Postgresql
+      Http
     ].freeze
   end
 end

integrations/spec/multiwoven/integrations/destination/http/client_spec.rb

@@ -0,0 +1,142 @@
+# frozen_string_literal: true
+
+RSpec.describe Multiwoven::Integrations::Destination::Http::Client do # rubocop:disable Metrics/BlockLength
+  include WebMock::API
+
+  before(:each) do
+    WebMock.disable_net_connect!(allow_localhost: true)
+  end
+
+  let(:client) { described_class.new }
+  let(:mock_http_session) { double("Net::Http::Session") }
+  let(:connection_config) do
+    {
+      destination_url: "https://www.google.com",
+      headers: {
+        test: "test",
+        test1: "test1"
+      }
+    }.with_indifferent_access
+  end
+
+  let(:sync_config_json) do
+    { source: {
+        name: "DestinationConnectorName",
+        type: "destination",
+        connection_specification: {
+          private_api_key: "test_api_key"
+        }
+      },
+      destination: {
+        name: "Http",
+        type: "destination",
+        connection_specification: connection_config
+      },
+      model: {
+        name: "ExampleModel",
+        query: "SELECT * FROM CALL_CENTER LIMIT 1",
+        query_type: "raw_sql",
+        primary_key: "id"
+      },
+      sync_mode: "incremental",
+      destination_sync_mode: "insert",
+      stream: {
+        name: "test",
+        url: "test",
+        request_method: "POST",
+        json_schema: {
+          type: "object",
+          properties: {
+            name: {
+              type: %w[string null]
+            }
+          }
+        }
+      } }.with_indifferent_access
+  end
+
+  let(:records) do
+    [{ name: "John Doe" }]
+  end
+  let(:csv_content) { "id,name\n1,Test Record\n" }
+
+  describe "#check_connection" do
+    context "when the connection is successful" do
+      before do
+        stub_request(:post, "https://www.google.com")
+          .to_return(status: 200, body: "", headers: {})
+      end
+
+      it "returns a successful connection status" do
+        response = client.check_connection(connection_config)
+        expect(response).to be_a(Multiwoven::Integrations::Protocol::MultiwovenMessage)
+        expect(response.connection_status.status).to eq("succeeded")
+      end
+    end
+
+    context "when the connection fails" do
+      before do
+        stub_request(:post, "https://www.google.com")
+          .to_return(status: 404, body: "", headers: {})
+      end
+
+      it "returns a failed connection status with an error message" do
+        response = client.check_connection(connection_config)
+
+        expect(response).to be_a(Multiwoven::Integrations::Protocol::MultiwovenMessage)
+        expect(response.connection_status.status).to eq("failed")
+      end
+    end
+  end
+
+  describe "#discover" do
+    it "returns a catalog" do
+      message = subject.discover
+      catalog = message.catalog
+      expect(catalog).to be_a(Multiwoven::Integrations::Protocol::Catalog)
+      expect(catalog.request_rate_limit).to eql(600)
+      expect(catalog.request_rate_limit_unit).to eql("minute")
+      expect(catalog.schema_mode).to eql("schemaless")
+    end
+  end
+
+  describe "#write" do
+    context "when the write operation is successful" do
+      before do
+        stub_request(:post, "https://www.google.com")
+          .to_return(status: 200, body: "", headers: {})
+      end
+
+      it "increments the success count" do
+        sync_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json(
+          sync_config_json.to_json
+        )
+        response = client.write(sync_config, records)
+        expect(response.tracking.success).to eq(records.size)
+        expect(response.tracking.failed).to eq(0)
+      end
+    end
+
+    context "when the write operation fails" do
+      before do
+        stub_request(:post, "https://www.google.com")
+          .to_return(status: 400, body: "", headers: {})
+      end
+
+      it "increments the failure count" do
+        sync_config = Multiwoven::Integrations::Protocol::SyncConfig.from_json(
+          sync_config_json.to_json
+        )
+        response = client.write(sync_config, records)
+        expect(response.tracking.failed).to eq(records.size)
+        expect(response.tracking.success).to eq(0)
+      end
+    end
+  end
+
+  def sync_config
+    Multiwoven::Integrations::Protocol::SyncConfig.from_json(
+      sync_config_json.to_json
+    )
+  end
+end