perf: Validate records with fastjsonschema

pyproject.toml

@@ -43,6 +43,7 @@ backoff = { version = ">=2.0.0", python = "<4" }
 backports-datetime-fromisoformat = { version = ">=2.0.1", python = "<3.11" }
 click = "~=8.0"
 cryptography = ">=3.4.6"
+fastjsonschema = ">=2.19.0"
 fs = ">=2.4.16"
 importlib-metadata = {version = "<7.0.0", python = "<3.12"}
 importlib-resources = {version = ">=5.12.0", markers = "python_version < \"3.9\""}
@@ -216,6 +217,7 @@ warn_unused_ignores = true
 ignore_missing_imports = true
 module = [
     "backports.datetime_fromisoformat.*",
+    "fastjsonschema.*",
     "joblib.*",  # TODO: Remove when https://github.com/joblib/joblib/issues/1516 is shipped
     "jsonpath_ng.*",
     "pyarrow.*", # TODO: Remove when https://github.com/apache/arrow/issues/32609 if implemented and released

singer_sdk/exceptions.py

@@ -137,4 +137,12 @@ class ConformedNameClashException(Exception):
 
 
 class MissingKeyPropertiesError(Exception):
-    """Raised when a recieved (and/or transformed) record is missing key properties."""
+    """Raised when a received (and/or transformed) record is missing key properties."""
+
+
+class InvalidJSONSchema(Exception):
+    """Raised when a JSON schema is invalid."""
+
+
+class InvalidRecord(Exception):
+    """Raised when a stream record is invalid according to its declared schema."""

singer_sdk/io_base.py

@@ -25,10 +25,10 @@ class SingerReader(metaclass=abc.ABCMeta):
     def listen(self, file_input: t.IO[str] | None = None) -> None:
         """Read from input until all messages are processed.
 
+        This method is internal to the SDK and should not need to be overridden.
+
         Args:
             file_input: Readable stream of messages. Defaults to standard in.
-
-        This method is internal to the SDK and should not need to be overridden.
         """
         if not file_input:
             file_input = sys.stdin

singer_sdk/sinks/core.py

@@ -13,9 +13,14 @@
 from gzip import open as gzip_open
 from types import MappingProxyType
 
-from jsonschema import Draft7Validator
+import fastjsonschema
+from typing_extensions import override
 
-from singer_sdk.exceptions import MissingKeyPropertiesError
+from singer_sdk.exceptions import (
+    InvalidJSONSchema,
+    InvalidRecord,
+    MissingKeyPropertiesError,
+)
 from singer_sdk.helpers._batch import (
     BaseBatchFileEncoding,
     BatchConfig,
@@ -39,7 +44,76 @@
 
     from singer_sdk.target_base import Target
 
-JSONSchemaValidator = Draft7Validator
+
+class BaseJSONSchemaValidator(abc.ABC):
+    """Abstract base class for JSONSchema validator."""
+
+    def __init__(self, schema: dict[str, t.Any]) -> None:
+        """Initialize the record validator.
+
+        Args:
+            schema: Schema of the stream to sink.
+        """
+        self.schema = schema
+
+    @abc.abstractmethod
+    def validate(self, record: dict[str, t.Any]) -> None:
+        """Validate a record message.
+
+        This method MUST raise an ``InvalidRecord`` exception if the record is invalid.
+
+        Args:
+            record: Record message to validate.
+        """
+
+
+class FastJSONSchemaValidator(BaseJSONSchemaValidator):
+    """Validate records using the ``fastjsonschema`` library."""
+
+    def __init__(
+        self,
+        schema: dict,
+        *,
+        validate_formats: bool = False,
+        format_validators: dict[str, t.Callable] | None = None,
+    ):
+        """Initialize the validator.
+
+        Args:
+            schema: Schema of the stream to sink.
+            validate_formats: Whether JSON string formats (e.g. ``date-time``) should
+                be validated.
+            format_validators: User-defined format validators.
+
+        Raises:
+            InvalidJSONSchema: If the schema provided from tap or mapper is invalid.
+        """
+        super().__init__(schema)
+        try:
+            self.validator = fastjsonschema.compile(
+                self.schema,
+                use_formats=validate_formats,
+                formats=format_validators or {},
+            )
+        except fastjsonschema.JsonSchemaDefinitionException as e:
+            error_message = f"Schema Validation Error: {e}"
+            raise InvalidJSONSchema(error_message) from e
+
+    @override
+    def validate(self, record: dict):  # noqa: ANN201
+        """Validate a record message.
+
+        Args:
+            record: Record message to validate.
+
+        Raises:
+            InvalidRecord: If the record is invalid.
+        """
+        try:
+            self.validator(record)
+        except fastjsonschema.JsonSchemaValueException as e:
+            error_message = f"Record Message Validation Error: {e.message}"
+            raise InvalidRecord(error_message) from e
 
 
 class Sink(metaclass=abc.ABCMeta):
@@ -51,6 +125,15 @@ class Sink(metaclass=abc.ABCMeta):
 
     MAX_SIZE_DEFAULT = 10000
 
+    validate_schema = True
+    """Enable JSON schema record validation."""
+
+    validate_field_string_format = False
+    """Enable JSON schema format validation, for example `date-time` string fields."""
+
+    fail_on_record_validation_exception: bool = True
+    """Interrupt the target execution when a record fails schema validation."""
+
     def __init__(
         self,
         target: Target,
@@ -95,10 +178,24 @@ def __init__(
         self._batch_records_read: int = 0
         self._batch_dupe_records_merged: int = 0
 
-        self._validator = Draft7Validator(
-            schema,
-            format_checker=Draft7Validator.FORMAT_CHECKER,
-        )
+        self._validator: BaseJSONSchemaValidator | None = self.get_validator()
+
+    def get_validator(self) -> BaseJSONSchemaValidator | None:
+        """Get a record validator for this sink.
+
+        Override this method to use a custom format validator, or disable record
+        validation by returning `None`.
+
+        Returns:
+            An instance of a subclass of ``BaseJSONSchemaValidator``.
+        """
+        if self.validate_schema:
+            return FastJSONSchemaValidator(
+                self.schema,
+                validate_formats=self.validate_field_string_format,
+                format_validators={},
+            )
+        return None
 
     def _get_context(self, record: dict) -> dict:  # noqa: ARG002
         """Return an empty dictionary by default.
@@ -328,8 +425,20 @@ def _validate_and_parse(self, record: dict) -> dict:
 
         Returns:
             TODO
+
+        Raises:
+            InvalidRecord: If the record is invalid.
         """
-        self._validator.validate(record)
+        if self._validator is not None:
+            # TODO: Check the performance impact of this try/except block. It runs
+            # on every record, so it's probably bad and should be moved up the stack.
+            try:
+                self._validator.validate(record)
+            except InvalidRecord as e:
+                if self.fail_on_record_validation_exception:
+                    raise InvalidRecord(e) from e
+                self.logger.exception("Record validation failed %s", e)
+
         self._parse_timestamps_in_record(
             record=record,
             schema=self.schema,