Merge pull request #3 from linuxfoundation/dynamodb-no-validate

Add validate_records option to target-nats-kv

Author: emsearcy

charts/lfx-v1-sync-helper/values.yaml

@@ -7,7 +7,7 @@ replicas: 1
 # image is the configuration for the container images
 image:
   # repository is the container image repository
-  repository: ghcr.io/linuxfoundation/lfx-v1-sync-helper/v1-sync-helper
+  repository: ghcr.io/linuxfoundation/lfx-v1-sync-helper/lfx-v1-sync-helper
   # tag is the container image tag (overrides appVersion from Chart.yaml)
   tag: ""
   # pullPolicy is the image pull policy

meltano/load/target-nats-kv/src/target_nats_kv/__init__.py

@@ -53,6 +53,7 @@ async def persist_messages(
     kv_client: KeyValue,
     key_prefix: str,
     refresh_mode: str,
+    validate_records: bool = True,
 ) -> dict | None:
     """
     Process Singer messages.
@@ -114,10 +115,7 @@ async def persist_messages(
                 continue
             if primary_key_value[0] == "$":
                 logger.warning(
-                    (
-                        "Ignoring record for stream %s with "
-                        "primary key starting with $"
-                    ),
+                    "Ignoring record for stream %s with primary key starting with $",
                     stream,
                 )
                 continue
@@ -133,20 +131,21 @@ async def persist_messages(
                     )
                     continue
 
-            try:
-                validators[stream].validate(o.record)
-            except jsonschema.ValidationError as e:
-                logger.warning(
-                    (
-                        "Ignoring record %s for stream %s that fails "
-                        "schema validation: %s at .%s"
-                    ),
-                    primary_key_value,
-                    stream,
-                    e.message,
-                    ".".join(str(i) for i in e.relative_path),
-                )
-                continue
+            if validate_records:
+                try:
+                    validators[stream].validate(o.record)
+                except jsonschema.ValidationError as e:
+                    logger.warning(
+                        (
+                            "Ignoring record %s for stream %s that fails "
+                            "schema validation: %s at .%s"
+                        ),
+                        primary_key_value,
+                        stream,
+                        e.message,
+                        ".".join(str(i) for i in e.relative_path),
+                    )
+                    continue
 
             key = f"{key_prefix}{stream}.{primary_key_value}"
 
@@ -163,13 +162,9 @@ async def persist_messages(
                 bookmark = bookmarks[stream]
                 if bookmark is not None and len(bookmark) > 0:
                     bookmark_attr = bookmark[0]
-            if bookmark_attr is not None and refresh_mode != "full":
-                if bookmark_attr in o.record:
-                    bookmark_source = o.record[bookmark_attr]
-                else:
-                    bookmark_source = ""
-
-                # Compare data freshness using "bookmark" attribute.
+            if refresh_mode != "full":
+                # Compare data using either the "bookmark" or a data
+                # comparison.
                 try:
                     # Try to fetch current value.
                     current = await kv_client.get(key)
@@ -197,20 +192,48 @@ async def persist_messages(
                         )
                         continue
 
-                    if bookmark_attr in current_record:
-                        bookmark_target = current_record[bookmark_attr]
+                    should_update = False
+                    if bookmark_attr is not None:
+                        # Compare bookmark values as strings to determine if the
+                        # new record should be refreshed based on the refresh mode.
+                        # Note, "newer" (which is the default) is implicit in the
+                        # "or" clause of this condition (if it's not "full" or
+                        # "same", it must be "newer").
+                        if bookmark_attr in o.record:
+                            bookmark_source = o.record[bookmark_attr]
+                        else:
+                            bookmark_source = ""
+
+                        if bookmark_attr in current_record:
+                            bookmark_target = current_record[bookmark_attr]
+                        else:
+                            bookmark_target = ""
+
+                        if (
+                            refresh_mode == "same"
+                            and str(bookmark_source) >= str(bookmark_target)
+                        ) or str(bookmark_source) > str(bookmark_target):
+                            should_update = True
+                    elif refresh_mode == "same":
+                        # If there is no bookmark, "same" means always update.
+                        should_update = True
                     else:
-                        bookmark_target = ""
-
-                    # Compare bookmark values as strings to determine if the
-                    # new record should be refreshed based on the refresh mode.
-                    # Note, "newer" (which is the default) is implicit in the
-                    # "or" clause of this condition (if it's not "full" or
-                    # "same", it must be "newer").
-                    if (
-                        refresh_mode == "same"
-                        and str(bookmark_source) >= str(bookmark_target)
-                    ) or str(bookmark_source) > str(bookmark_target):
+                        # No bookmark, and refresh_mode=="newer" (by process of
+                        # elimination), so, compare the data itself.
+                        target_record_filtered = {
+                            k: v
+                            for k, v in current_record.items()
+                            if not k.startswith("_sdc_")
+                        }
+                        source_record_filtered = {
+                            k: v
+                            for k, v in o.record.items()
+                            if not k.startswith("_sdc_")
+                        }
+                        if target_record_filtered != source_record_filtered:
+                            should_update = True
+
+                    if should_update:
                         # Update with revision
                         await kv_client.update(
                             key=key,
@@ -222,7 +245,7 @@ async def persist_messages(
                         logger.debug(
                             (
                                 "Skipping record for stream %s with key %s due to "
-                                "not being newer."
+                                "not needing update."
                             ),
                             stream,
                             primary_key_value,
@@ -236,8 +259,8 @@ async def persist_messages(
                         value=json.dumps(o.record).encode("utf-8"),
                     )
             else:
-                # No bookmarks for this stream, or, user has requested "full"
-                # sync, so use regular put.
+                # User has requested "full" sync, so use "put" without
+                # data checks.
                 await kv_client.put(
                     key=key,
                     value=json.dumps(o.record).encode("utf-8"),
@@ -278,6 +301,7 @@ async def run(config: dict) -> dict | None:
     bucket = config.get("bucket", "singer")
     key_prefix = config.get("key_prefix", "")
     refresh_mode = config.get("refresh_mode", "")
+    validate_records = config.get("validate_records", True)
     if user_credentials is not None:
         user_credentials = Path(user_credentials)
 
@@ -298,6 +322,7 @@ async def run(config: dict) -> dict | None:
         kv_client,
         key_prefix,
         refresh_mode,
+        validate_records,
     )
     return state
 

meltano/meltano.yml

@@ -52,6 +52,28 @@ plugins:
           # No timestamp field for this table.
           replication-key: ""
       use_aws_env_vars: true
+    schema:
+      # Override auto-detected schema for alternate columns and datatypes that
+      # weren't seen in the sample data.
+      itx-zoom-meetings-registrants-v2:
+        occurrence:
+          type: ["string", "null"]
+      itx-zoom-meetings-v2:
+        # Matches both `.recurrence` and `.updated_occurrences[].recurrence`.
+        "*recurrence.monthly_day":
+          type: ["integer", "null"]
+        "*recurrence.end_times":
+          type: ["integer", "null"]
+        use_unique_ics_uid:
+          type: ["string", "null"]
+      itx-zoom-past-meetings:
+        "*recurrence.monthly_day":
+          type: ["integer", "null"]
+        "*recurrence.end_times":
+          type: ["integer", "null"]
+      itx-zoom-past-meetings-attendees:
+        is_auto_matched:
+          type: ["boolean", "null"]
   loaders:
   - name: target-nats-kv
     namespace: target_nats_kv
@@ -79,3 +101,6 @@ plugins:
       - label: Newer only
         value: newer
       value: newer
+    - name: validate_records
+      kind: boolean
+      value: true