Repository: amp-labs/connectors
Pull request #1076 (2024-10-15 11:56:10)

feat: Add Pipedrive Read Connector
This Adds the Pipedrive Read connector

Output Tests:
<img width="1198" alt="Screenshot 2024-10-15 at 14 55 14" src="https://github.com/user-attachments/assets/8011933e-2343-4cf0-9fea-859db26cd7a1">




Patch:
diff --git a/providers/pipedrive/README.md b/providers/pipedrive/README.md
new file mode 100644
index 00000000..92777e8b
--- /dev/null
+++ b/providers/pipedrive/README.md
@@ -0,0 +1,43 @@
+# Pipedrive connector
+
+
+## Read
+Read is used to list all records of a given type. For example, if you want to list all user activities, you would use the `Read` method with the `activities` object.
+
+## Supported Objects 
+Below is an exhaustive list of the supported Objects in the Pipedrive deep connector with their endpoint resources(ObjectName).
+
+Pipedrive API version : v1
+
+| Object | Resource | Method |
+| :-------- | :------- | :-------- |
+| Activities | activities | Read |
+| Activity Fields | activityFields | Read |
+| Activity Types | activityTypes | Read |
+| Call Logs | callLogs | Read |
+| Currencies | currencies| Read |
+| Deals | deals | Read |
+| Deal Fields | dealFields| Read |
+| Files | files | Read |
+| Filters | filters | Read |
+| Leads | leads| Read |
+| Lead Labels | leadLabels | Read |
+| Lead Sources | leadSources | Read |
+| Legacy Teams | legacyTeams | Read |
+| Notes | notes | Read |
+| Note Fields | noteFields | Read |
+| Organizations | organizations | Read |
+| Organization Fields | organizationFields | Read, Write |
+| Organization Relationships | organizationRelationships | Read |
+| Permission Sets | permissionSets | Read |
+| Persons |  persons | Read |
+| Person Fields | personFields | Read |
+| Pipelines  | pipelines | Read |
+| Products | products | Read |
+| Product Fields | productFields | Read |
+| Roles | roles | Read | 
+| Stages | stages | Read |
+| Users | users | Read |
+| User Connections | userConnections | Read |
+| User Settings | userSettings | Read |
+| Webhooks | webhooks | Read |
\ No newline at end of file
diff --git a/providers/pipedrive/connector.go b/providers/pipedrive/connector.go
index 7f05f4a4..7b3a18a8 100644
--- a/providers/pipedrive/connector.go
+++ b/providers/pipedrive/connector.go
@@ -62,6 +62,6 @@ func (c *Connector) setBaseURL(newURL string) {
 	c.Client.HTTPClient.Base = newURL
 }
 
-func (c *Connector) getApiURL(arg string) (*urlbuilder.URL, error) {
+func (c *Connector) getAPIURL(arg string) (*urlbuilder.URL, error) {
 	return urlbuilder.New(c.BaseURL, apiVersion, arg)
 }
diff --git a/providers/pipedrive/metadata.go b/providers/pipedrive/metadata.go
index 251eb54a..33abfe94 100644
--- a/providers/pipedrive/metadata.go
+++ b/providers/pipedrive/metadata.go
@@ -29,7 +29,7 @@ func (c *Connector) ListObjectMetadata(ctx context.Context,
 
 	for _, obj := range objectNames {
 		// Constructing the  request url.
-		url, err := c.getApiURL(obj)
+		url, err := c.getAPIURL(obj)
 		if err != nil {
 			return nil, err
 		}
diff --git a/providers/pipedrive/objectNames.go b/providers/pipedrive/objectNames.go
new file mode 100644
index 00000000..3fc468b3
--- /dev/null
+++ b/providers/pipedrive/objectNames.go
@@ -0,0 +1,12 @@
+package pipedrive
+
+import (
+	"github.com/amp-labs/connectors/common/handy"
+)
+
+// ObjectNameToResponseField maps ObjectName to the response field name which contains that object.
+var ObjectNameToResponseField = handy.NewDefaultMap(map[string]string{}, //nolint:gochecknoglobals
+	func(key string) string {
+		return key
+	},
+)
diff --git a/providers/pipedrive/parse.go b/providers/pipedrive/parse.go
new file mode 100644
index 00000000..50a10c32
--- /dev/null
+++ b/providers/pipedrive/parse.go
@@ -0,0 +1,34 @@
+package pipedrive
+
+import (
+	"strconv"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/jsonquery"
+	"github.com/amp-labs/connectors/common/urlbuilder"
+	"github.com/spyzhov/ajson"
+)
+
+// nextRecordsURL builds the next-page url func.
+func nextRecordsURL(url *urlbuilder.URL) common.NextPageFunc {
+	return func(node *ajson.Node) (string, error) {
+		// check if there is more items in the collection.
+		more, err := jsonquery.New(node, "additional_data", "pagination").Bool("more_items_in_collection", true)
+		if err != nil {
+			return "", err
+		}
+
+		startValue, err := jsonquery.New(node, "additional_data", "pagination").Integer("next_start", true)
+		if err != nil {
+			return "", err
+		}
+
+		if *more {
+			url.WithQueryParam("start", strconv.FormatInt(*startValue, 10))
+
+			return url.String(), nil
+		}
+
+		return "", nil
+	}
+}
diff --git a/providers/pipedrive/read.go b/providers/pipedrive/read.go
new file mode 100644
index 00000000..a0de2e15
--- /dev/null
+++ b/providers/pipedrive/read.go
@@ -0,0 +1,54 @@
+package pipedrive
+
+import (
+	"context"
+	"time"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/urlbuilder"
+)
+
+func (c *Connector) Read(ctx context.Context, config common.ReadParams) (*common.ReadResult, error) {
+	if err := config.ValidateParams(true); err != nil {
+		return nil, err
+	}
+
+	url, err := c.buildReadURL(config)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := c.Client.Get(ctx, url.String())
+	if err != nil {
+		return nil, err
+	}
+
+	return common.ParseResult(resp,
+		common.GetRecordsUnderJSONPath("data"),
+		nextRecordsURL(url),
+		common.GetMarshaledData,
+		config.Fields,
+	)
+}
+
+func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, error) {
+	// If NextPage is set, then we're reading the next page of results.
+	// The NextPage URL has all the necessary parameters.
+	if len(config.NextPage) > 0 {
+		return urlbuilder.New(config.NextPage.String())
+	}
+
+	url, err := c.getAPIURL(config.ObjectName)
+	if err != nil {
+		return nil, err
+	}
+
+	// begin fetching objects at provided start date
+	// Supporting objects are: Activities & Notes only.
+	if !config.Since.IsZero() {
+		startDate := config.Since.Format(time.DateOnly)
+		url.WithQueryParam("start_date", startDate)
+	}
+
+	return url, nil
+}


Test patch:
diff --git a/providers/pipedrive/read_test.go b/providers/pipedrive/read_test.go
new file mode 100644
index 00000000..4753d50f
--- /dev/null
+++ b/providers/pipedrive/read_test.go
@@ -0,0 +1,210 @@
+package pipedrive
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/test/utils/mockutils"
+	"github.com/amp-labs/connectors/test/utils/mockutils/mockserver"
+	"github.com/amp-labs/connectors/test/utils/testroutines"
+	"github.com/amp-labs/connectors/test/utils/testutils"
+)
+
+func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
+	t.Parallel()
+
+	nextPageTest := testutils.DataFromFile(t, "activities.json")
+	leads := testutils.DataFromFile(t, "leads.json")
+
+	ErrResponseBody := `{
+		"success":false,
+		"error":"Scope and URL mismatch",
+		"errorCode":403,
+		"error_info":"Please check developers.pipedrive.com"
+	}`
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(nextPageTest)
+	}))
+
+	tests := []testroutines.Read{
+		{
+			Name:         "Object Name Required",
+			Server:       mockserver.Dummy(),
+			ExpectedErrs: []error{common.ErrMissingObjects},
+		},
+		{
+			Name:         "At Least One Read Field Required",
+			Input:        common.ReadParams{ObjectName: "activities"},
+			Server:       mockserver.Dummy(),
+			ExpectedErrs: []error{common.ErrMissingFields},
+		},
+		{
+			Name:         "NextPage URL construction",
+			Input:        common.ReadParams{ObjectName: "activities", Fields: connectors.Fields("id")},
+			Server:       server,
+			ExpectedErrs: nil,
+			Expected: &common.ReadResult{
+				Rows: 1,
+				Data: []common.ReadResultRow{
+					{
+						Fields: map[string]any{
+							"id": float64(2),
+						},
+						Raw: map[string]any{
+							"id":                            float64(2),
+							"company_id":                    float64(13313052),
+							"user_id":                       float64(20580207),
+							"done":                          false,
+							"type":                          "call",
+							"reference_type":                nil,
+							"reference_id":                  nil,
+							"conference_meeting_client":     nil,
+							"conference_meeting_url":        nil,
+							"due_date":                      "2024-10-30",
+							"due_time":                      "",
+							"duration":                      "",
+							"busy_flag":                     false,
+							"add_time":                      "2024-10-16 12:16:02",
+							"marked_as_done_time":           "",
+							"last_notification_time":        nil,
+							"last_notification_user_id":     nil,
+							"notification_language_id":      nil,
+							"subject":                       "I usually can't come up with words",
+							"public_description":            "Demo activity",
+							"calendar_sync_include_context": nil,
+							"location":                      "Dar es salaam",
+							"org_id":                        nil,
+							"person_id":                     nil,
+							"deal_id":                       nil,
+							"lead_id":                       nil,
+							"active_flag":                   true,
+							"update_time":                   "2024-10-16 12:16:02",
+							"update_user_id":                nil,
+							"source_timezone":               nil,
+							"rec_rule":                      nil,
+							"rec_rule_extension":            nil,
+							"rec_master_activity_id":        nil,
+							"conference_meeting_id":         nil,
+							"original_start_time":           nil,
+							"private":                       false,
+							"priority":                      nil,
+							"note":                          nil,
+							"created_by_user_id":            float64(20580207),
+							"location_subpremise":           nil,
+							"location_street_number":        nil,
+							"location_route":                nil,
+							"location_sublocality":          nil,
+							"location_locality":             nil,
+							"location_admin_area_level_1":   nil,
+							"location_admin_area_level_2":   nil,
+							"location_country":              nil,
+							"location_postal_code":          nil,
+							"location_formatted_address":    nil,
+							"attendees":                     nil,
+							"participants":                  nil,
+							"series":                        nil,
+							"is_recurring":                  nil,
+							"org_name":                      nil,
+							"person_name":                   nil,
+							"deal_title":                    nil,
+							"lead_title":                    nil,
+							"owner_name":                    "Integration User",
+							"person_dropbox_bcc":            nil,
+							"deal_dropbox_bcc":              nil,
+							"assigned_to_user_id":           float64(20580207),
+							"type_name":                     "Call",
+							"lead":                          nil,
+						},
+					},
+				},
+				Done:     false,
+				NextPage: common.NextPageToken(fmt.Sprintf("%s/v1/activities?start=1", server.URL)),
+			},
+		},
+		{
+			Name:  "Not Found Resource or Higher Suite Resource",
+			Input: common.ReadParams{ObjectName: "activitiess", Fields: connectors.Fields("id")},
+			Server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusForbidden)
+				mockutils.WriteBody(w, ErrResponseBody)
+			})),
+			ExpectedErrs: []error{common.NewHTTPStatusError(http.StatusForbidden,
+				fmt.Errorf("%w: %s", common.ErrForbidden, ErrResponseBody))},
+			Expected: nil,
+		},
+		{
+			Name: "Successful Read",
+			Input: common.ReadParams{
+				ObjectName: "leads",
+				Fields:     connectors.Fields("channel", "id", "origin", "title"),
+			},
+			Server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write(leads)
+			})),
+			Comparator: func(baseURL string, actual, expected *common.ReadResult) bool {
+				// custom comparison focuses on subset of fields to keep the test short
+				return mockutils.ReadResultComparator.SubsetFields(actual, expected) &&
+					mockutils.ReadResultComparator.SubsetRaw(actual, expected) &&
+					actual.NextPage.String() == expected.NextPage.String() &&
+					actual.Done == expected.Done
+			},
+			Expected: &common.ReadResult{
+				Data: []common.ReadResultRow{{
+					Fields: map[string]any{
+						"channel": nil,
+						"id":      "8ba414d0-8ad8-11ef-9e9e-07879dd74146",
+						"origin":  "ManuallyCreated",
+						"title":   "Ampersand lead",
+					},
+					Raw: map[string]any{
+						"add_time":            "2024-10-15T09:33:33.213Z",
+						"cc_email":            "integrationuser-sandbox2+13313052+leadif7ZKQ7mhWz28LivD9BmE3@pipedrivemail.com",
+						"channel":             nil,
+						"channel_id":          nil,
+						"creator_id":          float64(20580207),
+						"expected_close_date": nil,
+						"id":                  "8ba414d0-8ad8-11ef-9e9e-07879dd74146",
+						"is_archived":         false,
+						"label_ids":           []any{},
+						"next_activity_id":    nil,
+						"organization_id":     float64(2),
+						"origin":              "ManuallyCreated",
+						"origin_id":           nil,
+						"owner_id":            float64(20580207),
+						"person_id":           float64(2),
+						"source_name":         "Manually created",
+						"title":               "Ampersand lead",
+						"update_time":         "2024-10-15T09:33:33.213Z",
+						"value":               nil,
+						"visible_to":          "3",
+						"was_seen":            true,
+					},
+				}},
+				Done: true,
+			},
+			ExpectedErrs: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		// nolint:varnamelen
+		tt := tt // rebind, omit loop side effects for parallel goroutine
+		t.Run(tt.Name, func(t *testing.T) {
+			t.Parallel()
+
+			tt.Run(t, func() (connectors.ReadConnector, error) {
+				return constructTestConnector(tt.Server.URL)
+			})
+		})
+	}
+}
diff --git a/providers/pipedrive/test/activities.json b/providers/pipedrive/test/activities.json
new file mode 100644
index 00000000..d838f150
--- /dev/null
+++ b/providers/pipedrive/test/activities.json
@@ -0,0 +1,90 @@
+{
+    "success": true,
+    "data": [
+        {
+            "id": 2,
+            "company_id": 13313052,
+            "user_id": 20580207,
+            "done": false,
+            "type": "call",
+            "reference_type": null,
+            "reference_id": null,
+            "conference_meeting_client": null,
+            "conference_meeting_url": null,
+            "due_date": "2024-10-30",
+            "due_time": "",
+            "duration": "",
+            "busy_flag": false,
+            "add_time": "2024-10-16 12:16:02",
+            "marked_as_done_time": "",
+            "last_notification_time": null,
+            "last_notification_user_id": null,
+            "notification_language_id": null,
+            "subject": "I usually can't come up with words",
+            "public_description": "Demo activity",
+            "calendar_sync_include_context": null,
+            "location": "Dar es salaam",
+            "org_id": null,
+            "person_id": null,
+            "deal_id": null,
+            "lead_id": null,
+            "active_flag": true,
+            "update_time": "2024-10-16 12:16:02",
+            "update_user_id": null,
+            "source_timezone": null,
+            "rec_rule": null,
+            "rec_rule_extension": null,
+            "rec_master_activity_id": null,
+            "conference_meeting_id": null,
+            "original_start_time": null,
+            "private": false,
+            "priority": null,
+            "note": null,
+            "created_by_user_id": 20580207,
+            "location_subpremise": null,
+            "location_street_number": null,
+            "location_route": null,
+            "location_sublocality": null,
+            "location_locality": null,
+            "location_admin_area_level_1": null,
+            "location_admin_area_level_2": null,
+            "location_country": null,
+            "location_postal_code": null,
+            "location_formatted_address": null,
+            "attendees": null,
+            "participants": null,
+            "series": null,
+            "is_recurring": null,
+            "org_name": null,
+            "person_name": null,
+            "deal_title": null,
+            "lead_title": null,
+            "owner_name": "Integration User",
+            "person_dropbox_bcc": null,
+            "deal_dropbox_bcc": null,
+            "assigned_to_user_id": 20580207,
+            "type_name": "Call",
+            "lead": null
+        }
+    ],
+    "additional_data": {
+        "pagination": {
+            "start": 0,
+            "limit": 1,
+            "more_items_in_collection": true,
+            "next_start": 1
+        }
+    },
+    "related_objects": {
+        "user": {
+            "20580207": {
+                "id": 20580207,
+                "name": "Integration User",
+                "email": "integration.user+pipedrive@withampersand.com",
+                "has_pic": 0,
+                "pic_hash": null,
+                "active_flag": true
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/providers/pipedrive/test/leads.json b/providers/pipedrive/test/leads.json
new file mode 100644
index 00000000..77b25f16
--- /dev/null
+++ b/providers/pipedrive/test/leads.json
@@ -0,0 +1,35 @@
+{
+    "success": true,
+    "data": [
+        {
+            "id": "8ba414d0-8ad8-11ef-9e9e-07879dd74146",
+            "title": "Ampersand lead",
+            "owner_id": 20580207,
+            "creator_id": 20580207,
+            "label_ids": [],
+            "value": null,
+            "expected_close_date": null,
+            "person_id": 2,
+            "organization_id": 2,
+            "is_archived": false,
+            "source_name": "Manually created",
+            "origin": "ManuallyCreated",
+            "origin_id": null,
+            "channel": null,
+            "channel_id": null,
+            "was_seen": true,
+            "next_activity_id": null,
+            "add_time": "2024-10-15T09:33:33.213Z",
+            "update_time": "2024-10-15T09:33:33.213Z",
+            "visible_to": "3",
+            "cc_email": "integrationuser-sandbox2+13313052+leadif7ZKQ7mhWz28LivD9BmE3@pipedrivemail.com"
+        }
+    ],
+    "additional_data": {
+        "pagination": {
+            "start": 0,
+            "limit": 100,
+            "more_items_in_collection": false
+        }
+    }
+}
\ No newline at end of file
diff --git a/test/pipedrive/read/read.go b/test/pipedrive/read/read.go
new file mode 100644
index 00000000..9af835aa
--- /dev/null
+++ b/test/pipedrive/read/read.go
@@ -0,0 +1,112 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/test/pipedrive"
+	"github.com/amp-labs/connectors/test/utils"
+)
+
+func main() {
+	// Handle Ctrl-C gracefully.
+	ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer done()
+
+	// Set up slog logging.
+	utils.SetupLogging()
+
+	conn := pipedrive.GetPipedriveConnector(ctx)
+	defer utils.Close(conn)
+
+	if err := readActivities(ctx, conn); err != nil {
+		slog.Error(err.Error())
+	}
+
+	if err := readDeals(ctx, conn); err != nil {
+		slog.Error(err.Error())
+	}
+
+	if err := readLeads(ctx, conn); err != nil {
+		slog.Error(err.Error())
+	}
+
+}
+
+func readActivities(ctx context.Context, conn connectors.ReadConnector) error {
+	config := connectors.ReadParams{
+		ObjectName: "activities",
+		Since:      time.Now().Add(-720 * time.Hour),
+		Fields:     connectors.Fields("user_id", "done", "id"),
+	}
+
+	result, err := conn.Read(ctx, config)
+	if err != nil {
+		return err
+	}
+
+	// Print the results
+	jsonStr, err := json.MarshalIndent(result, "", "  ")
+	if err != nil {
+		return err
+	}
+
+	_, _ = os.Stdout.Write(jsonStr)
+	_, _ = os.Stdout.WriteString("\n")
+
+	return nil
+}
+
+func readDeals(ctx context.Context, conn connectors.ReadConnector) error {
+	config := connectors.ReadParams{
+		ObjectName: "deals",
+		Since:      time.Now().Add(-720 * time.Hour),
+		Fields:     connectors.Fields("close_time", "id"),
+	}
+
+	result, err := conn.Read(ctx, config)
+	if err != nil {
+		return err
+	}
+
+	// Print the results
+	jsonStr, err := json.MarshalIndent(result, "", "  ")
+	if err != nil {
+		return err
+	}
+
+	_, _ = os.Stdout.Write(jsonStr)
+	_, _ = os.Stdout.WriteString("\n")
+
+	return nil
+}
+
+func readLeads(ctx context.Context, conn connectors.ReadConnector) error {
+	config := connectors.ReadParams{
+		ObjectName: "leads",
+		Since:      time.Now().Add(-720 * time.Hour),
+		Fields:     connectors.Fields("origin", "channel", "title", "id"),
+	}
+
+	result, err := conn.Read(ctx, config)
+	if err != nil {
+		return err
+	}
+
+	// Print the results
+	jsonStr, err := json.MarshalIndent(result, "", "  ")
+	if err != nil {
+		return err
+	}
+
+	_, _ = os.Stdout.Write(jsonStr)
+	_, _ = os.Stdout.WriteString("\n")
+
+	return nil
+}