feat: Transient streams that set up context for child streams

[ReubenFrankel]

Sometimes I find myself writing streams like

class _MessageMediumsStream(Stream):
    """Define message mediums stream."""

    name = "_message_mediums"
    schema = th.ObjectType().to_dict()
    selected = False  # use for context generation only

    @override
    def get_records(self, context):
        yield from ({"messageMedium": m} for m in ["Email", "Push", "InApp", "SMS"])

    @override
    def get_child_context(self, record, context):
        return record


class TemplatesStream(IterableStream):
    """Define templates stream."""

    parent_stream_type = _MessageMediumsStream
    name = "templates"
    path = "/templates"
    records_jsonpath = "$.templates[*]"
    schema_filepath = SCHEMAS_DIR / "templates.json"
    key_properties = ("templateId",)
    replication_key = "updatedAt"

    @override
    def get_url_params(self, context, next_page_token):
        params = super().get_url_params(context, next_page_token)
        params["messageMedium"] = context["messageMedium"]

        if start_date := self.get_starting_timestamp(context):
            params["startDateTime"] = start_date.strftime(r"%Y-%m-%d %H:%M:%S")

        return params

    @override
    def get_child_context(self, record, context):
        return {**context, "templateId": record["templateId"]}


class EmailTemplatesStream(IterableStream):
    """Define email templates stream."""

    parent_stream_type = TemplatesStream
    name = "email_templates"
    path = "/templates/email/get"
    schema_filepath = SCHEMAS_DIR / "email_templates.json"
    key_properties = ("templateId",)

    @override
    def get_records(self, context):
        if context["messageMedium"] != "Email":
            return

        yield from super().get_records(context)

    @override
    def get_url_params(self, context, next_page_token):
        params = super().get_url_params(context, next_page_token)
        params["templateId"] = context["templateId"]

so I can control how nested child streams run. Is there a better way to do this?

If not, it would be nicer to use something like

class _MessageMediumsStream(IterableStream):
    """Define message mediums stream."""

    transient = True

    @override
    def generate_child_contexts(self, record, context):
        yield from ({"messageMedium": m} for m in ["Email", "Push", "InApp", "SMS"])

rather than having to define a name, schema - which are currently required - and selected = False to "hide" the stream.

Another example that goes off and makes requests to setup context:

class _MetadataStream(IterableStream):
    """Define metadata stream."""

    name = "_metadata"
    path = "/metadata"
    schema = th.ObjectType().to_dict()
    selected = False  # use for context generation only
    records_jsonpath = "$.results[*]"

    @override
    def get_child_context(self, record, context):
        return {"table": record["name"]}


class _MetadataTablesStream(IterableStream):
    """Define metadata tables stream."""

    parent_stream_type = _MetadataStream
    name = "_metadata_tables"
    path = "/metadata/{table}"
    schema = th.ObjectType().to_dict()
    selected = False  # use for context generation only
    records_jsonpath = "$.results[*]"

    @override
    def get_child_context(self, record, context):
        return {**context, "key": record["key"]}


class MetadataStream(IterableStream):
    """Define metadata stream."""

    parent_stream_type = _MetadataTablesStream
    name = "metadata"
    path = "/metadata/{table}/{key}"
    schema_filepath = SCHEMAS_DIR / "metadata.json"
    key_properties = ("key",)