feat: Support union schemas

docs/classes/typing/singer_sdk.typing.Constant.rst

@@ -0,0 +1,8 @@
+singer_sdk.typing.Constant
+==========================
+
+.. currentmodule:: singer_sdk.typing
+
+.. autoclass:: Constant
+    :members:
+    :special-members: __init__, __call__

docs/classes/typing/singer_sdk.typing.DiscriminatedUnion.rst

@@ -0,0 +1,8 @@
+singer_sdk.typing.DiscriminatedUnion
+====================================
+
+.. currentmodule:: singer_sdk.typing
+
+.. autoclass:: DiscriminatedUnion
+    :members:
+    :special-members: __init__, __call__

docs/classes/typing/singer_sdk.typing.OneOf.rst

@@ -0,0 +1,8 @@
+singer_sdk.typing.OneOf
+=======================
+
+.. currentmodule:: singer_sdk.typing
+
+.. autoclass:: OneOf
+    :members:
+    :special-members: __init__, __call__

docs/reference.rst

@@ -92,9 +92,11 @@ JSON Schema builder classes
     typing.PropertiesList
     typing.ArrayType
     typing.BooleanType
+    typing.Constant
     typing.CustomType
     typing.DateTimeType
     typing.DateType
+    typing.DiscriminatedUnion
     typing.DurationType
     typing.EmailType
     typing.HostnameType
@@ -104,6 +106,7 @@ JSON Schema builder classes
     typing.JSONPointerType
     typing.NumberType
     typing.ObjectType
+    typing.OneOf
     typing.Property
     typing.RegexType
     typing.RelativeJSONPointerType

singer_sdk/typing.py

@@ -711,6 +711,150 @@ def type_dict(self) -> dict:  # type: ignore[override]
         return result
 
 
+class OneOf(JSONPointerType):
+    """OneOf type.
+
+    This type allows for a value to be one of a set of types.
+
+    Examples:
+        >>> t = OneOf(StringType, IntegerType)
+        >>> print(t.to_json(indent=2))
+        {
+            "oneOf": [
+                {
+                    "type": [
+                        "string"
+                    ]
+                },
+                {
+                    "type": [
+                        "integer"
+                    ]
+                }
+            ]
+        }
+    """
+
+    def __init__(self, *types: W | type[W]) -> None:
+        """Initialize OneOf type.
+
+        Args:
+            types: Types to choose from.
+        """
+        self.wrapped = types
+
+    @property
+    def type_dict(self) -> dict:  # type: ignore[override]
+        """Get type dictionary.
+
+        Returns:
+            A dictionary describing the type.
+        """
+        return {"oneOf": [t.type_dict for t in self.wrapped]}
+
+
+class Constant(JSONTypeHelper):
+    """A constant property.
+
+    A property that is always the same value.
+
+    Examples:
+        >>> t = Constant("foo")
+        >>> print(t.to_json(indent=2))
+        {
+            "const": "foo"
+        }
+    """
+
+    def __init__(self, value: _JsonValue) -> None:
+        """Initialize Constant.
+
+        Args:
+            value: Value of the constant.
+        """
+        self.value = value
+
+    @property
+    def type_dict(self) -> dict:  # type: ignore[override]
+        """Get type dictionary.
+
+        Returns:
+            A dictionary describing the type.
+        """
+        return {"const": self.value}
+
+
+class DiscriminatedUnion(OneOf):
+    """A discriminator property.
+
+    This is a special case of :class:`singer_sdk.typing.OneOf`, where values are
+    JSON objects, and the type of the object is determined by a property in the
+    object.
+
+    The property is a :class:`singer_sdk.typing.Constant` called the discriminator
+    property.
+    """
+
+    def __init__(self, key: str, **options: ObjectType) -> None:
+        """Initialize a discriminated union type.
+
+        Args:
+            key: Name of the discriminator property.
+            options: Mapping of discriminator values to object types.
+
+        Examples:
+            >>> t = DiscriminatedUnion("species", cat=ObjectType(), dog=ObjectType())
+            >>> print(t.to_json(indent=2))
+            {
+              "oneOf": [
+                {
+                  "type": "object",
+                  "properties": {
+                    "species": {
+                      "const": "cat",
+                      "description": "Discriminator for object of type 'cat'."
+                    }
+                  },
+                  "required": [
+                    "species"
+                  ]
+                },
+                {
+                  "type": "object",
+                  "properties": {
+                    "species": {
+                        "const": "dog",
+                        "description": "Discriminator for object of type 'dog'."
+                    }
+                  },
+                  "required": [
+                    "species"
+                  ]
+                }
+              ]
+            }
+        """
+        self.key = key
+        self.options = options
+
+        super().__init__(
+            *(
+                ObjectType(
+                    Property(
+                        key,
+                        Constant(k),
+                        required=True,
+                        description=f"Discriminator for object of type '{k}'.",
+                    ),
+                    *v.wrapped.values(),
+                    additional_properties=v.additional_properties,
+                    pattern_properties=v.pattern_properties,
+                )
+                for k, v in options.items()
+            ),
+        )
+
+
 class CustomType(JSONTypeHelper):
     """Accepts an arbitrary JSON Schema dictionary."""
 

tests/core/test_jsonschema_helpers.py

@@ -7,6 +7,7 @@
 from textwrap import dedent
 
 import pytest
+from jsonschema import Draft6Validator
 
 from singer_sdk.helpers._typing import (
     JSONSCHEMA_ANNOTATION_SECRET,
@@ -30,6 +31,7 @@
     CustomType,
     DateTimeType,
     DateType,
+    DiscriminatedUnion,
     DurationType,
     EmailType,
     HostnameType,
@@ -813,3 +815,48 @@ def test_type_check_variations(property_schemas, type_check_functions, results):
     for property_schema in property_schemas:
         for type_check_function, result in zip(type_check_functions, results):
             assert type_check_function(property_schema) == result
+
+
+def test_discriminated_union():
+    th = DiscriminatedUnion(
+        "flow",
+        oauth=ObjectType(
+            Property("client_id", StringType, required=True, secret=True),
+            Property("client_secret", StringType, required=True, secret=True),
+            additional_properties=False,
+        ),
+        password=ObjectType(
+            Property("username", StringType, required=True),
+            Property("password", StringType, required=True, secret=True),
+            additional_properties=False,
+        ),
+    )
+
+    validator = Draft6Validator(th.to_dict())
+
+    assert validator.is_valid(
+        {
+            "flow": "oauth",
+            "client_id": "123",
+            "client_secret": "456",
+        },
+    )
+    assert validator.is_valid(
+        {
+            "flow": "password",
+            "password": "123",
+            "username": "456",
+        },
+    )
+    assert not validator.is_valid(
+        {
+            "flow": "oauth",
+            "client_id": "123",
+        },
+    )
+    assert not validator.is_valid(
+        {
+            "flow": "password",
+            "client_id": "123",
+        },
+    )