From 77c3f90a0cd20e58daf494ac129905ed64a89944 Mon Sep 17 00:00:00 2001
From: Ken Payne <ken@meltano.com>
Date: Wed, 5 Jul 2023 12:27:38 +0100
Subject: [PATCH] feat: Validate parsed/transformed record against schema
 message (#1769)

Co-authored-by: Edgar R. M <edgar@meltano.com>
---
 singer_sdk/exceptions.py           |  4 ++++
 singer_sdk/sinks/core.py           | 20 ++++++++++++++++++++
 singer_sdk/target_base.py          |  3 ++-
 singer_sdk/testing/target_tests.py | 10 +++++++++-
 4 files changed, 35 insertions(+), 2 deletions(-)

diff --git a/singer_sdk/exceptions.py b/singer_sdk/exceptions.py
index 221a26f46..23325aa2a 100644
--- a/singer_sdk/exceptions.py
+++ b/singer_sdk/exceptions.py
@@ -115,3 +115,7 @@ class ConformedNameClashException(Exception):
 
     e.g. two columns conformed to the same name
     """
+
+
+class MissingKeyPropertiesError(Exception):
+    """Raised when a recieved (and/or transformed) record is missing key properties."""
diff --git a/singer_sdk/sinks/core.py b/singer_sdk/sinks/core.py
index 2510be03c..ec43c4060 100644
--- a/singer_sdk/sinks/core.py
+++ b/singer_sdk/sinks/core.py
@@ -15,6 +15,7 @@
 from dateutil import parser
 from jsonschema import Draft7Validator, FormatChecker
 
+from singer_sdk.exceptions import MissingKeyPropertiesError
 from singer_sdk.helpers._batch import (
     BaseBatchFileEncoding,
     BatchConfig,
@@ -321,6 +322,25 @@ def _validate_and_parse(self, record: dict) -> dict:
         )
         return record
 
+    def _singer_validate_message(self, record: dict) -> None:
+        """Ensure record conforms to Singer Spec.
+
+        Args:
+            record: Record (after parsing, schema validations and transformations).
+
+        Raises:
+            MissingKeyPropertiesError: If record is missing one or more key properties.
+        """
+        if not all(key_property in record for key_property in self.key_properties):
+            msg = (
+                f"Record is missing one or more key_properties. \n"
+                f"Key Properties: {self.key_properties}, "
+                f"Record Keys: {list(record.keys())}"
+            )
+            raise MissingKeyPropertiesError(
+                msg,
+            )
+
     def _parse_timestamps_in_record(
         self,
         record: dict,
diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py
index facd69c87..7a1be112c 100644
--- a/singer_sdk/target_base.py
+++ b/singer_sdk/target_base.py
@@ -338,9 +338,10 @@ def _process_record_message(self, message_dict: dict) -> None:
                 sink._remove_sdc_metadata_from_record(transformed_record)
 
             sink._validate_and_parse(transformed_record)
+            transformed_record = sink.preprocess_record(transformed_record, context)
+            sink._singer_validate_message(transformed_record)
 
             sink.tally_record_read()
-            transformed_record = sink.preprocess_record(transformed_record, context)
             sink.process_record(transformed_record, context)
             sink._after_process_record(context)
 
diff --git a/singer_sdk/testing/target_tests.py b/singer_sdk/testing/target_tests.py
index e1bd6662b..8412329c5 100644
--- a/singer_sdk/testing/target_tests.py
+++ b/singer_sdk/testing/target_tests.py
@@ -4,7 +4,10 @@
 
 import pytest
 
-from singer_sdk.exceptions import RecordsWithoutSchemaException
+from singer_sdk.exceptions import (
+    MissingKeyPropertiesError,
+    RecordsWithoutSchemaException,
+)
 
 from .templates import TargetFileTestTemplate, TargetTestTemplate
 
@@ -108,6 +111,11 @@ class TargetRecordMissingKeyProperty(TargetFileTestTemplate):
 
     name = "record_missing_key_property"
 
+    def test(self) -> None:
+        """Run test."""
+        with pytest.raises(MissingKeyPropertiesError):
+            super().test()
+
 
 class TargetRecordMissingRequiredProperty(TargetFileTestTemplate):
     """Test Target handles record missing required property."""