feat(shopify): Incremental updates

plugins/source/shopify/client/client.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/cloudquery/cloudquery/plugins/source/shopify/internal/shopify"
+	"github.com/cloudquery/plugin-sdk/backend"
 	"github.com/cloudquery/plugin-sdk/plugins/source"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
@@ -21,14 +22,16 @@ type Client struct {
 	spSpec     Spec
 
 	Services *shopify.Client
+	Backend  backend.Backend
 }
 
-func New(logger zerolog.Logger, sourceSpec specs.Source, spSpec Spec, services *shopify.Client) Client {
+func New(logger zerolog.Logger, sourceSpec specs.Source, spSpec Spec, services *shopify.Client, bk backend.Backend) Client {
 	return Client{
 		logger:     logger,
 		sourceSpec: sourceSpec,
 		spSpec:     spSpec,
 		Services:   services,
+		Backend:    bk,
 	}
 }
 
@@ -40,7 +43,7 @@ func (c *Client) ID() string {
 	return c.sourceSpec.Name
 }
 
-func Configure(ctx context.Context, logger zerolog.Logger, s specs.Source, _ source.Options) (schema.ClientMeta, error) {
+func Configure(ctx context.Context, logger zerolog.Logger, s specs.Source, o source.Options) (schema.ClientMeta, error) {
 	spSpec := &Spec{}
 	if err := s.UnmarshalSpec(spSpec); err != nil {
 		return nil, fmt.Errorf("failed to unmarshal shopify spec: %w", err)
@@ -51,7 +54,7 @@ func Configure(ctx context.Context, logger zerolog.Logger, s specs.Source, _ sou
 		return nil, err
 	}
 
-	cl := New(logger, s, *spSpec, services)
+	cl := New(logger, s, *spSpec, services, o.Backend)
 	return &cl, nil
 }
 
@@ -72,16 +75,20 @@ func getServiceClient(logger zerolog.Logger, spec *Spec) (*shopify.Client, error
 	if spec.MaxRetries < 1 {
 		spec.MaxRetries = 30
 	}
+	if spec.PageSize < 1 {
+		spec.PageSize = 50
+	}
 
-	return shopify.New(
-		logger,
-		&http.Client{
+	return shopify.New(shopify.ClientOptions{
+		Log: logger,
+		HC: &http.Client{
 			Timeout: time.Duration(spec.Timeout) * time.Second,
 		},
-		spec.APIKey,
-		spec.APISecret,
-		spec.AccessToken,
-		spec.ShopURL,
-		spec.MaxRetries,
-	)
+		ApiKey:      spec.APIKey,
+		ApiSecret:   spec.APISecret,
+		AccessToken: spec.AccessToken,
+		ShopURL:     spec.ShopURL,
+		MaxRetries:  spec.MaxRetries,
+		PageSize:    int(spec.PageSize),
+	})
 }

plugins/source/shopify/client/spec.go

@@ -13,4 +13,5 @@ type Spec struct {
 	// Optional
 	Timeout    int64 `json:"timeout_secs,omitempty"`
 	MaxRetries int64 `json:"max_retries,omitempty"`
+	PageSize   int64 `json:"page_size,omitempty"`
 }

plugins/source/shopify/client/testing.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/cloudquery/cloudquery/plugins/source/shopify/internal/shopify"
+	"github.com/cloudquery/plugin-sdk/backend"
 	"github.com/cloudquery/plugin-sdk/plugins/source"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
@@ -22,6 +23,10 @@ import (
 
 const testToken = "SomeToken"
 
+type TestOptions struct {
+	Backend backend.Backend
+}
+
 type MockHttpClient struct {
 	rootURL string
 	scheme  string
@@ -52,7 +57,7 @@ func (c *MockHttpClient) Do(req *http.Request) (*http.Response, error) {
 	return c.client.Do(req)
 }
 
-func MockTestHelper(t *testing.T, table *schema.Table, createServices func(*mux.Router) error) {
+func MockTestHelper(t *testing.T, table *schema.Table, createServices func(*mux.Router) error, opts TestOptions) {
 	version := "vDev"
 
 	t.Helper()
@@ -82,12 +87,19 @@ func MockTestHelper(t *testing.T, table *schema.Table, createServices func(*mux.
 			return nil, err
 		}
 
-		services, err := shopify.New(logger, mockClient, "", "", testToken, h.URL, 1)
+		services, err := shopify.New(shopify.ClientOptions{
+			Log:         logger,
+			HC:          mockClient,
+			AccessToken: testToken,
+			ShopURL:     h.URL,
+			MaxRetries:  1,
+			PageSize:    50,
+		})
 		if err != nil {
 			return nil, err
 		}
 
-		c := New(logger, spec, spSpec, services)
+		c := New(logger, spec, spSpec, services, opts.Backend)
 		return &c, nil
 	}
 

plugins/source/shopify/docs/tables/README.md

@@ -2,11 +2,11 @@
 
 ## Tables
 
-- [shopify_abandoned_checkouts](shopify_abandoned_checkouts.md)
-- [shopify_customers](shopify_customers.md)
-- [shopify_orders](shopify_orders.md)
-- [shopify_price_rules](shopify_price_rules.md)
+- [shopify_abandoned_checkouts](shopify_abandoned_checkouts.md) (Incremental)
+- [shopify_customers](shopify_customers.md) (Incremental)
+- [shopify_orders](shopify_orders.md) (Incremental)
+- [shopify_price_rules](shopify_price_rules.md) (Incremental)
   - [shopify_price_rule_discount_codes](shopify_price_rule_discount_codes.md)
-- [shopify_products](shopify_products.md)
+- [shopify_products](shopify_products.md) (Incremental)
   - [shopify_product_images](shopify_product_images.md)
   - [shopify_product_variants](shopify_product_variants.md)

plugins/source/shopify/docs/tables/shopify_abandoned_checkouts.md

@@ -1,6 +1,7 @@
 # Table: shopify_abandoned_checkouts
 
 The primary key for this table is **id**.
+It supports incremental syncs based on the **updated_at** column.
 
 ## Columns
 
@@ -11,13 +12,13 @@ The primary key for this table is **id**.
 |_cq_id|UUID|
 |_cq_parent_id|UUID|
 |id (PK)|Int|
+|updated_at (Incremental Key)|Timestamp|
 |token|String|
 |cart_token|String|
 |email|String|
 |gateway|String|
 |buyer_accepts_marketing|Bool|
 |created_at|Timestamp|
-|updated_at|Timestamp|
 |landing_site|String|
 |note_attributes|JSON|
 |referring_site|String|

plugins/source/shopify/docs/tables/shopify_customers.md

@@ -1,6 +1,7 @@
 # Table: shopify_customers
 
 The primary key for this table is **id**.
+It supports incremental syncs based on the **updated_at** column.
 
 ## Columns
 
@@ -11,10 +12,10 @@ The primary key for this table is **id**.
 |_cq_id|UUID|
 |_cq_parent_id|UUID|
 |id (PK)|Int|
+|updated_at (Incremental Key)|Timestamp|
 |email|String|
 |accepts_marketing|Bool|
 |created_at|Timestamp|
-|updated_at|Timestamp|
 |first_name|String|
 |last_name|String|
 |orders_count|Int|

plugins/source/shopify/docs/tables/shopify_orders.md

@@ -1,6 +1,7 @@
 # Table: shopify_orders
 
 The primary key for this table is **id**.
+It supports incremental syncs based on the (**created_at**, **updated_at**) columns.
 
 ## Columns
 
@@ -11,6 +12,8 @@ The primary key for this table is **id**.
 |_cq_id|UUID|
 |_cq_parent_id|UUID|
 |id (PK)|Int|
+|created_at (Incremental Key)|Timestamp|
+|updated_at (Incremental Key)|Timestamp|
 |admin_graphql_api_id|String|
 |app_id|Int|
 |browser_ip|String|
@@ -22,7 +25,6 @@ The primary key for this table is **id**.
 |closed_at|Timestamp|
 |confirmed|Bool|
 |contact_email|String|
-|created_at|Timestamp|
 |currency|String|
 |current_subtotal_price|String|
 |current_total_discounts|String|
@@ -61,7 +63,6 @@ The primary key for this table is **id**.
 |total_tax|String|
 |total_tip_received|String|
 |total_weight|Int|
-|updated_at|Timestamp|
 |user_id|Int|
 |customer|JSON|
 |discount_applications|JSON|

plugins/source/shopify/docs/tables/shopify_price_rules.md

@@ -1,7 +1,7 @@
 # Table: shopify_price_rules
 
 The primary key for this table is **id**.
-
+It supports incremental syncs based on the **updated_at** column.
 ## Relations
 
 The following tables depend on shopify_price_rules:
@@ -16,6 +16,7 @@ The following tables depend on shopify_price_rules:
 |_cq_id|UUID|
 |_cq_parent_id|UUID|
 |id (PK)|Int|
+|updated_at (Incremental Key)|Timestamp|
 |value_type|String|
 |value|String|
 |customer_selection|String|
@@ -27,7 +28,6 @@ The following tables depend on shopify_price_rules:
 |starts_at|Timestamp|
 |ends_at|Timestamp|
 |created_at|Timestamp|
-|updated_at|Timestamp|
 |entitled_product_ids|JSON|
 |entitled_variant_ids|JSON|
 |entitled_collection_ids|JSON|

plugins/source/shopify/docs/tables/shopify_products.md

@@ -1,7 +1,7 @@
 # Table: shopify_products
 
 The primary key for this table is **id**.
-
+It supports incremental syncs based on the **updated_at** column.
 ## Relations
 
 The following tables depend on shopify_products:
@@ -17,13 +17,13 @@ The following tables depend on shopify_products:
 |_cq_id|UUID|
 |_cq_parent_id|UUID|
 |id (PK)|Int|
+|updated_at (Incremental Key)|Timestamp|
 |title|String|
 |body_html|String|
 |vendor|String|
 |product_type|String|
 |created_at|Timestamp|
 |handle|String|
-|updated_at|Timestamp|
 |published_at|Timestamp|
 |template_suffix|String|
 |status|String|