perf: Validate records with fastjsonschema

pyproject.toml

@@ -44,6 +44,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", python = "<3.9"}
@@ -224,6 +225,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/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,7 +13,7 @@
 from gzip import open as gzip_open
 from types import MappingProxyType
 
-import jsonschema
+import fastjsonschema
 from typing_extensions import override
 
 from singer_sdk.exceptions import (
@@ -67,46 +67,36 @@ def validate(self, record: dict[str, t.Any]) -> None:
         """
 
 
-class JSONSchemaValidator(BaseJSONSchemaValidator):
+class FastJSONSchemaValidator(BaseJSONSchemaValidator):
     """Validate records using the ``fastjsonschema`` library."""
 
     def __init__(
         self,
         schema: dict,
         *,
         validate_formats: bool = False,
-        format_checker: jsonschema.FormatChecker | None = None,
+        format_checker: 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_checker: User-defined format checker.
+            format_checker: User-defined format validators.
 
         Raises:
             InvalidJSONSchema: If the schema provided from tap or mapper is invalid.
         """
-        jsonschema_validator = jsonschema.Draft7Validator
-
         super().__init__(schema)
-        if validate_formats:
-            format_checker = format_checker or jsonschema_validator.FORMAT_CHECKER
-        else:
-            format_checker = jsonschema.FormatChecker(formats=())
-
         try:
-            jsonschema_validator.check_schema(schema)
-        except jsonschema.SchemaError as e:
+            self.validator = fastjsonschema.compile(
+                self.schema, use_formats=validate_formats, formats=format_checker or {}
+            )
+        except fastjsonschema.JsonSchemaDefinitionException as e:
             error_message = f"Schema Validation Error: {e}"
             raise InvalidJSONSchema(error_message) from e
 
-        self.validator = jsonschema_validator(
-            schema=schema,
-            format_checker=format_checker,
-        )
-
     @override
     def validate(self, record: dict):  # noqa: ANN201
         """Validate a record message.
@@ -118,9 +108,10 @@ def validate(self, record: dict):  # noqa: ANN201
             InvalidRecord: If the record is invalid.
         """
         try:
-            self.validator.validate(record)
-        except jsonschema.ValidationError as e:
-            raise InvalidRecord(e.message, record) from e
+            self.validator(record)
+        except fastjsonschema.JsonSchemaValueException as e:
+            error_message = f"Record Message Validation Error: {e.message}"
+            raise InvalidRecord(error_message, record) from e
 
 
 class Sink(metaclass=abc.ABCMeta):
@@ -197,7 +188,7 @@ def get_validator(self) -> BaseJSONSchemaValidator | None:
             An instance of a subclass of ``BaseJSONSchemaValidator``.
         """
         if self.validate_schema:
-            return JSONSchemaValidator(
+            return FastJSONSchemaValidator(
                 self.schema,
                 validate_formats=self.validate_field_string_format,
             )

tests/core/sinks/test_type_checker.py

@@ -47,7 +47,7 @@ def process_record(self, record: dict, context: dict) -> None:
 
     with pytest.raises(
         InvalidJSONSchema,
-        match=r"Schema Validation Error: 'ssttrriinngg' is not valid under any",
+        match=r"Schema Validation Error: Unknown type: 'ssttrriinngg'",
     ):
         CustomSink(target, "test_stream", test_schema_invalid, None)