singer_sdk.schema.source.StreamSchema

class singer_sdk.schema.source.StreamSchema[source]

Stream schema descriptor.

Assign a StreamSchema descriptor to a stream class to dynamically load the schema from a schema source.

Example

class MyStream(Stream):

schema = StreamSchema(SchemaDirectory(“schemas”))

Example with OpenAPI:
class MyStream(Stream):

schema = StreamSchema(OpenAPISchema(“openapi.json”))

Example with custom OpenAPI preprocessor:
class MyStream(Stream):
schema = StreamSchema(
OpenAPISchema(

“openapi.json”, preprocessor=CustomPreprocessor(),

)

)

__init__(schema_source, *, key=None)[source]

Initialize the stream schema.

Parameters:

  • schema_source (SchemaSource[_TKey]) – The schema source to use.

  • key (_TKey | None) – The Optional key to use to get the schema from the schema source. by default the stream name will be used.

Return type:

None

get_stream_schema(stream, stream_class)[source]

Get the schema from the stream instance or class.

Parameters:

  • stream (_TStream) – The stream instance to get the schema from.

  • stream_class (type[_TStream]) – The stream class to get the schema from.

Returns:

A JSON schema dictionary.

Return type:

dict[str, Any]