Class CreateSource

misc/python/materialize/zippy/source_actions.py:89–118

class CreateSource(Action):
    def __init__(self, capabilities: Capabilities, source: SourceExists) -> None:
        self.source = source
        super().__init__(capabilities)

    def run(self, c: Composition, state: State) -> None:
        envelope = str(self.source.topic.envelope).split(".")[1]
        kafka_connection_name = f"{self.source.name}_kafka_conn"
        c.testdrive(
            dedent(f"""
                > CREATE CONNECTION IF NOT EXISTS {self.source.name}_csr_conn
                  TO CONFLUENT SCHEMA REGISTRY (URL '${{testdrive.schema-registry-url}}');

                > CREATE CONNECTION IF NOT EXISTS {kafka_connection_name}
                  TO KAFKA (BROKER '${{testdrive.kafka-addr}}' {'USING SSH TUNNEL zippy_ssh' if self.source.uses_ssh_tunnel else ''}, SECURITY PROTOCOL PLAINTEXT);

                > CREATE SOURCE {self.source.name}
                  IN CLUSTER {self.source.cluster_name}
                  FROM KAFKA CONNECTION {kafka_connection_name}
                  (TOPIC 'testdrive-{self.source.topic.name}-${{testdrive.seed}}')

                > CREATE TABLE {self.source.get_name_for_query()} FROM SOURCE {self.source.name} (REFERENCE "testdrive-{self.source.topic.name}-${{testdrive.seed}}")
                  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION {self.source.name}_csr_conn
                  ENVELOPE {envelope}
                """),
            mz_service=state.mz_service,
        )

    def provides(self) -> list[Capability]:
        return [self.source]


class AlterSourceConnection(Action):