Repository: amp-labs/connectors
Pull request #1890 (2025-07-21 17:12:57)

[CON-2024] feat(stripe): Incremental read
# Live Tests
### Invoices
<img width="1268" height="443" alt="image" src="https://github.com/user-attachments/assets/fa3e4202-1701-45ba-ab61-6e1b1aa97a3f" />



Cobalt0s: * **#1890** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1890?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a> 👈 <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1890?utm_source=stack-comment-view-in-graphite" target="_blank">(View in Graphite)</a>
* `main`


<h2></h2>

This stack of pull requests is managed by <a href="https://graphite.dev?utm-source=stack-comment"><b>Graphite</b></a>. Learn more about <a href="https://stacking.dev/?utm_source=stack-comment">stacking</a>.
<!-- Current dependencies on/for this PR: -->

Patch:
diff --git a/common/types.go b/common/types.go
index d3179f9e..07cfc3fa 100644
--- a/common/types.go
+++ b/common/types.go
@@ -262,7 +262,9 @@ type Association struct {
 	// ObjectID is the ID of the associated object.
 	ObjectId string `json:"objectId"`
 	// AssociationType is the type of association.
-	AssociationType string `json:"associationType,omitempty"`
+	AssociationType string         `json:"associationType,omitempty"`
+	Raw             map[string]any `json:"raw,omitempty"`
+	Fields          map[string]any `json:"fields,omitempty"`
 }
 
 // WriteResult is what's returned from writing data via the Write call.
diff --git a/providers/braze/handlers.go b/providers/braze/handlers.go
index 7b3c12df..9f8ed4a7 100644
--- a/providers/braze/handlers.go
+++ b/providers/braze/handlers.go
@@ -20,7 +20,15 @@ type catalogPayload struct {
 var ErrInvalidData = errors.New("invalid request data provided")
 
 func (c *Connector) metadataRequest(ctx context.Context, objectName string) (*http.Request, error) {
-	url, err := urlbuilder.New(c.ProviderInfo().BaseURL, objectName)
+	endpoint := objectName
+
+	// map objectName to the braze APIs endpoints.
+	if objectName, exists := readEndpointsByObject[objectName]; exists {
+		// map the objectName to the appropriate endpoint
+		endpoint = objectName
+	}
+
+	url, err := urlbuilder.New(c.ProviderInfo().BaseURL, endpoint)
 	if err != nil {
 		return nil, err
 	}
@@ -51,6 +59,12 @@ func (c *Connector) parseMetadataResponse(
 		return nil, err
 	}
 
+	// map objectName to the braze APIs endpoints.
+	if endpoint, exists := readEndpointsByObject[objectName]; exists {
+		// map the objectName to the appropriate endpoint
+		objectName = endpoint
+	}
+
 	fld := dataFields.Get(objectName)
 
 	rcds, okay := (*data)[fld].([]any)
@@ -80,6 +94,12 @@ func (c *Connector) constructReadURL(params common.ReadParams) (*urlbuilder.URL,
 		return urlbuilder.New(params.NextPage.String())
 	}
 
+	// map objectName to the braze APIs endpoints.
+	if objectName, exists := readEndpointsByObject[params.ObjectName]; exists {
+		// map the objectName to the appropriate endpoint
+		params.ObjectName = objectName
+	}
+
 	url, err := urlbuilder.New(c.ProviderInfo().BaseURL, params.ObjectName)
 	if err != nil {
 		return nil, err
@@ -98,6 +118,12 @@ func (c *Connector) buildReadRequest(ctx context.Context, params common.ReadPara
 		return nil, err
 	}
 
+	// map objectName to the braze APIs endpoints.
+	if objectName, exists := readEndpointsByObject[params.ObjectName]; exists {
+		// map the objectName to the appropriate endpoint
+		params.ObjectName = objectName
+	}
+
 	if err := setSinceQuery(params, url); err != nil {
 		return nil, err
 	}
@@ -116,6 +142,12 @@ func (c *Connector) parseReadResponse(
 		return nil, err
 	}
 
+	// map objectName to the braze APIs endpoints.
+	if objectName, exists := readEndpointsByObject[params.ObjectName]; exists {
+		// map the objectName to the appropriate endpoint
+		params.ObjectName = objectName
+	}
+
 	return common.ParseResult(response,
 		common.ExtractRecordsFromPath(dataFields.Get(params.ObjectName)),
 		getNextRecordsURL(params.ObjectName, url, response),
@@ -152,6 +184,12 @@ func (c *Connector) buildWriteRequest(ctx context.Context, params common.WritePa
 		method   = http.MethodPost
 	)
 
+	// map objectName to the braze APIs endpoints.
+	if objectName, exists := writeEndpointsByObject[params.ObjectName]; exists {
+		// map the objectName to the appropriate endpoint
+		params.ObjectName = objectName
+	}
+
 	url, err = urlbuilder.New(c.ProviderInfo().BaseURL, params.ObjectName)
 	if err != nil {
 		return nil, err
diff --git a/providers/braze/support.go b/providers/braze/support.go
index 3edc434f..17d27676 100644
--- a/providers/braze/support.go
+++ b/providers/braze/support.go
@@ -25,3 +25,20 @@ func supportedOperations() components.EndpointRegistryInput {
 		},
 	}
 }
+
+var readEndpointsByObject = map[string]string{ //nolint: gochecknoglobals
+	"campaigns":           "campaigns/list",
+	"canvas":              "canvas/list",
+	"segments":            "segments/list",
+	"preference_center":   "preference_center/v1/list",
+	"subscription/status": "subscription/status/get",
+	"content_blocks":      "content_blocks/list",
+	"templates/email":     "templates/email/list",
+}
+
+var writeEndpointsByObject = map[string]string{ //nolint: gochecknoglobals
+	"subscription/status": "subscription/status/set",
+	"preference_center":   "preference_center/v1",
+	"content_blocks":      "content_blocks/create",
+	"templates/email":     "templates/email/create",
+}
diff --git a/providers/stripe/read.go b/providers/stripe/read.go
index 85c20639..0639fb7d 100644
--- a/providers/stripe/read.go
+++ b/providers/stripe/read.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/common/urlbuilder"
+	"github.com/amp-labs/connectors/internal/datautils"
 	"github.com/amp-labs/connectors/providers/stripe/metadata"
 )
 
@@ -41,28 +42,32 @@ func (c *Connector) Read(ctx context.Context, config common.ReadParams) (*common
 	)
 }
 
-func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, error) {
-	if len(config.NextPage) != 0 {
+func (c *Connector) buildReadURL(params common.ReadParams) (*urlbuilder.URL, error) {
+	if len(params.NextPage) != 0 {
 		// Next page
-		return urlbuilder.New(config.NextPage.String())
+		return urlbuilder.New(params.NextPage.String())
 	}
 
 	// First page
-	url, err := c.getURL(config.ObjectName)
+	url, err := c.getURL(params.ObjectName)
 	if err != nil {
 		return nil, err
 	}
 
 	url.WithQueryParam("limit", strconv.Itoa(DefaultPageSize))
 
+	if !params.Since.IsZero() && incrementalObjects.Has(params.ObjectName) {
+		url.WithQueryParam("created[gte]", strconv.FormatInt(params.Since.Unix(), 10))
+	}
+
 	// Deeply nested objects can be requested as part of a single API request.
 	// Example: Query parameter "expand[]=data.customer" will expand the nested customer object.
 	//
 	// For more details, refer to the Stripe documentation:
 	// https://docs.stripe.com/expand#how-it-works
-	if len(config.AssociatedObjects) != 0 {
-		expandTargets := make([]string, len(config.AssociatedObjects))
-		for index, associate := range config.AssociatedObjects {
+	if len(params.AssociatedObjects) != 0 {
+		expandTargets := make([]string, len(params.AssociatedObjects))
+		for index, associate := range params.AssociatedObjects {
 			expandTargets[index] = "data." + associate
 		}
 
@@ -71,3 +76,48 @@ func (c *Connector) buildReadURL(config common.ReadParams) (*urlbuilder.URL, err
 
 	return url, nil
 }
+
+var incrementalObjects = datautils.NewSet( // nolint:gochecknoglobals
+	"accounts",
+	"application_fees",
+	"balance/history",
+	"balance_transactions",
+	"charges",
+	"checkout/sessions",
+	"coupons",
+	"credit_notes",
+	"customers",
+	"disputes",
+	"events",
+	"file_links",
+	"files",
+	"forwarding/requests",
+	"identity/verification_reports",
+	"identity/verification_sessions",
+	"invoiceitems",
+	"invoices",
+	"issuing/authorizations",
+	"issuing/cardholders",
+	"issuing/cards",
+	"issuing/disputes",
+	"issuing/transactions",
+	"payment_intents",
+	"payouts",
+	"plans",
+	"prices",
+	"products",
+	"promotion_codes",
+	"radar/early_fraud_warnings",
+	"radar/value_lists",
+	"refunds",
+	"reporting/report_runs",
+	"reviews",
+	"setup_intents",
+	"shipping_rates",
+	"subscription_schedules",
+	"subscriptions",
+	"tax_rates",
+	"topups",
+	"transfers",
+	"treasury/financial_accounts",
+)


Test patch:
diff --git a/providers/stripe/read_test.go b/providers/stripe/read_test.go
index 0e0f1365..2893a408 100644
--- a/providers/stripe/read_test.go
+++ b/providers/stripe/read_test.go
@@ -4,6 +4,7 @@ import (
 	"errors"
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/common"
@@ -22,6 +23,7 @@ func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 	responseCustomersFirstPage := testutils.DataFromFile(t, "read/customers/1-first-page.json")
 	responseCustomersLastPage := testutils.DataFromFile(t, "read/customers/2-last-page.json")
 	responsePaymentsExpandedCustomer := testutils.DataFromFile(t, "read/payment_intents/expand_customer.json")
+	responseInvoices := testutils.DataFromFile(t, "read/invoices/incremental.json")
 
 	tests := []testroutines.Read{
 		{
@@ -167,6 +169,43 @@ func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 			},
 			ExpectedErrs: nil,
 		},
+		{
+			Name: "Incremental read of Invoices",
+			Input: common.ReadParams{
+				ObjectName: "invoices",
+				Fields:     connectors.Fields("description"),
+				Since:      time.Unix(1753116395, 0),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/v1/invoices"),
+					mockcond.QueryParam("limit", "100"),
+					mockcond.QueryParam("created[gte]", "1753116395"),
+				},
+				Then: mockserver.Response(http.StatusOK, responseInvoices),
+			}.Server(),
+			Comparator: testroutines.ComparatorSubsetRead,
+			Expected: &common.ReadResult{
+				Rows: 1,
+				Data: []common.ReadResultRow{{
+					Fields: map[string]any{
+						"description": "Invoice 8887. Potatoes",
+					},
+					Raw: map[string]any{
+						"billing_reason":    "manual",
+						"collection_method": "charge_automatically",
+						"customer":          "cus_Sbim60412VKvja",
+						"customer_email":    "freddy.buckley@company.com",
+						"customer_name":     "Freddy Buckley",
+					},
+				}},
+				NextPage: testroutines.URLTestServer + "/v1/invoices?" +
+					"created[gte]=1753116395&limit=100&starting_after=in_1RnN00ES6gLOjP91auKbmxwS",
+				Done: false,
+			},
+			ExpectedErrs: nil,
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/providers/stripe/test/read/invoices/incremental.json b/providers/stripe/test/read/invoices/incremental.json
new file mode 100644
index 00000000..3479c953
--- /dev/null
+++ b/providers/stripe/test/read/invoices/incremental.json
@@ -0,0 +1,127 @@
+{
+  "object": "list",
+  "data": [
+    {
+      "id": "in_1RnN00ES6gLOjP91auKbmxwS",
+      "object": "invoice",
+      "account_country": "US",
+      "account_name": "Ampersand-2",
+      "account_tax_ids": null,
+      "amount_due": 0,
+      "amount_overpaid": 0,
+      "amount_paid": 0,
+      "amount_remaining": 0,
+      "amount_shipping": 0,
+      "application": null,
+      "application_fee_amount": null,
+      "attempt_count": 0,
+      "attempted": false,
+      "auto_advance": false,
+      "automatic_tax": {
+        "disabled_reason": null,
+        "enabled": false,
+        "liability": null,
+        "provider": null,
+        "status": null
+      },
+      "automatically_finalizes_at": null,
+      "billing_reason": "manual",
+      "charge": null,
+      "collection_method": "charge_automatically",
+      "created": 1753116936,
+      "currency": "usd",
+      "custom_fields": null,
+      "customer": "cus_Sbim60412VKvja",
+      "customer_address": null,
+      "customer_email": "freddy.buckley@company.com",
+      "customer_name": "Freddy Buckley",
+      "customer_phone": null,
+      "customer_shipping": null,
+      "customer_tax_exempt": "none",
+      "customer_tax_ids": [],
+      "default_payment_method": null,
+      "default_source": null,
+      "default_tax_rates": [],
+      "description": "Invoice 8887. Potatoes",
+      "discount": null,
+      "discounts": [],
+      "due_date": null,
+      "effective_at": null,
+      "ending_balance": null,
+      "footer": null,
+      "from_invoice": null,
+      "hosted_invoice_url": null,
+      "invoice_pdf": null,
+      "issuer": {
+        "type": "self"
+      },
+      "last_finalization_error": null,
+      "latest_revision": null,
+      "lines": {
+        "object": "list",
+        "data": [],
+        "has_more": false,
+        "total_count": 0,
+        "url": "/v1/invoices/in_1RnN00ES6gLOjP91auKbmxwS/lines"
+      },
+      "livemode": false,
+      "metadata": {},
+      "next_payment_attempt": null,
+      "number": null,
+      "on_behalf_of": null,
+      "paid": false,
+      "paid_out_of_band": false,
+      "parent": null,
+      "payment_intent": null,
+      "payment_settings": {
+        "default_mandate": null,
+        "payment_method_options": null,
+        "payment_method_types": null
+      },
+      "period_end": 1753116936,
+      "period_start": 1753116936,
+      "post_payment_credit_notes_amount": 0,
+      "pre_payment_credit_notes_amount": 0,
+      "quote": null,
+      "receipt_number": null,
+      "rendering": {
+        "amount_tax_display": null,
+        "pdf": {
+          "page_size": "auto"
+        },
+        "template": null,
+        "template_version": null
+      },
+      "rendering_options": null,
+      "shipping_cost": null,
+      "shipping_details": null,
+      "starting_balance": 0,
+      "statement_descriptor": null,
+      "status": "draft",
+      "status_transitions": {
+        "finalized_at": null,
+        "marked_uncollectible_at": null,
+        "paid_at": null,
+        "voided_at": null
+      },
+      "subscription": null,
+      "subscription_details": {
+        "metadata": null
+      },
+      "subtotal": 0,
+      "subtotal_excluding_tax": 0,
+      "tax": null,
+      "test_clock": null,
+      "total": 0,
+      "total_discount_amounts": [],
+      "total_excluding_tax": 0,
+      "total_pretax_credit_amounts": [],
+      "total_tax_amounts": [],
+      "total_taxes": [],
+      "transfer_data": null,
+      "webhooks_delivered_at": 1753116936
+    }
+  ],
+  "has_more": true,
+  "url": "/v1/invoices"
+}
diff --git a/test/braze/metadata/main.go b/test/braze/metadata/main.go
index b12d8e54..a89886ad 100644
--- a/test/braze/metadata/main.go
+++ b/test/braze/metadata/main.go
@@ -14,7 +14,7 @@ func main() {
 
 	conn := braze.NewBrazeConnector(ctx)
 
-	m, err := conn.ListObjectMetadata(ctx, []string{"catalogs", "campaigns/list", "templates/email/list"})
+	m, err := conn.ListObjectMetadata(ctx, []string{"catalogs", "campaigns", "templates/email"})
 
 	if err != nil {
 		log.Fatal(err)
diff --git a/test/braze/read/main.go b/test/braze/read/main.go
index f3c769ec..c8cf6377 100644
--- a/test/braze/read/main.go
+++ b/test/braze/read/main.go
@@ -43,7 +43,7 @@ func main() {
 
 func testReadCampaigns(ctx context.Context, conn *br.Connector) error {
 	params := common.ReadParams{
-		ObjectName: "campaigns/list",
+		ObjectName: "campaigns",
 		Fields:     connectors.Fields("id", "name"),
 		Since:      time.Now().Add(-1 * time.Hour),
 		// NextPage:   "https://rest.iad-03.braze.com/campaigns/list?page=1",
@@ -92,7 +92,7 @@ func testReadEvents(ctx context.Context, conn *br.Connector) error {
 
 func testReadEmailTemplates(ctx context.Context, conn *br.Connector) error {
 	params := common.ReadParams{
-		ObjectName: "templates/email/list",
+		ObjectName: "templates/email",
 		Fields:     connectors.Fields("template_name", "email_template_id"),
 		Since:      time.Now().Add(-1 * time.Hour),
 		// NextPage:   "https://rest.iad-03.braze.com/templates/email/list?offset=101",
diff --git a/test/stripe/read/customers/main.go b/test/stripe/read/customers/main.go
index fa54561c..2cf8bb23 100644
--- a/test/stripe/read/customers/main.go
+++ b/test/stripe/read/customers/main.go
@@ -28,9 +28,9 @@ func main() {
 		Fields:     connectors.Fields("email", "name"),
 	})
 	if err != nil {
-		utils.Fail("error reading from Stripe", "error", err)
+		utils.Fail("error reading from provider", "error", err)
 	}
 
-	slog.Info("Reading customers..")
+	slog.Info("Reading...")
 	utils.DumpJSON(res, os.Stdout)
 }
diff --git a/test/stripe/read/invoices/main.go b/test/stripe/read/invoices/main.go
new file mode 100644
index 00000000..f9a9b3a4
--- /dev/null
+++ b/test/stripe/read/invoices/main.go
@@ -0,0 +1,38 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/common"
+	connTest "github.com/amp-labs/connectors/test/stripe"
+	"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 := connTest.GetStripeConnector(ctx)
+
+	res, err := conn.Read(ctx, common.ReadParams{
+		ObjectName: "invoices",
+		Fields:     connectors.Fields("description", "customer_name"),
+		Since:      time.Unix(1753116936, 0),
+	})
+	if err != nil {
+		utils.Fail("error reading from provider", "error", err)
+	}
+
+	slog.Info("Reading...")
+	utils.DumpJSON(res, os.Stdout)
+}