Typed Sinks

# sinks/publish.py
from providers.destination_client import DestinationClient
from sqlbuild.sinks import (
    LifecycleEvent,
    LifecycleEventKind,
    lifecycle_event_sink,
    lifecycle_event_to_json,
)


@lifecycle_event_sink(
    event_kinds={
        LifecycleEventKind.INVOCATION,
        LifecycleEventKind.RUN,
        LifecycleEventKind.RESOURCE,
        LifecycleEventKind.AUDIT,
    }
)
def publish_lifecycle(event: LifecycleEvent, destination_client: DestinationClient) -> None:
    destination_client.publish(
        route="sqlbuild.lifecycle.v2",
        key=event.invocation_id,
        payload=lifecycle_event_to_json(event).encode("utf-8"),
    )

from sqlbuild.sinks import CommandOutputRecord, command_output_sink, command_output_to_json


@command_output_sink(streams={"stdout", "stderr"})
def publish_output(record: CommandOutputRecord, destination_client: DestinationClient) -> None:
    destination_client.publish(
        route="sqlbuild.command_output.v1",
        key=record.record_id,
        payload=command_output_to_json(record).encode("utf-8"),
    )

[sinks.lifecycle]
event_kinds = ["run", "resource", "operation", "statement", "audit"]
min_severity = "info"

[sinks.lifecycle.named.publish_lifecycle]
event_kinds = ["resource", "statement"]
min_severity = "warning"