Function defs

misc/python/materialize/mzexplore/clone.py:27–166

Generate a DDL script that clones object_ids and dependencies that live in the same cluster into a different cluster.

(
    ddl_out: TextIO,
    cmp_out: TextIO,
    database: str,
    schema: str,
    cluster: str,
    object_ids: list[str],
    db_port: int,
    db_host: str,
    db_user: str,
    db_pass: str | None,
    db_require_ssl: bool,
    mzfmt: bool,
)

def defs(
    ddl_out: TextIO,
    cmp_out: TextIO,
    database: str,
    schema: str,
    cluster: str,
    object_ids: list[str],
    db_port: int,
    db_host: str,
    db_user: str,
    db_pass: str | None,
    db_require_ssl: bool,
    mzfmt: bool,
) -> None:
    """
    Generate a DDL script that clones object_ids and dependencies that live in
    the same cluster into a different cluster.
    """

    database_new = sql.identifier(database, True)
    schema_new = sql.identifier(schema, True)
    cluster_new = sql.identifier(cluster, True)

    with closing(
        sql.Database(
            port=db_port,
            host=db_host,
            user=db_user,
            database=None,
            password=db_pass,
            require_ssl=db_require_ssl,
        )
    ) as db:
        output_template = string.Template(textwrap.dedent("""
                -- original id: $id
                $create_sql
                """).lstrip())

        # Extract dependencies
        # --------------------

        if len(object_ids) == 0:
            msg = "At least one object_id necessary"
            raise ValueError(msg)

        old_clusters = list(db.object_clusters(object_ids))
        if len(old_clusters) != 1:
            msg = f"Cannot find unique old cluster for object ids: {object_ids}"
            raise ValueError(msg)

        [cluster_old] = old_clusters

        if cluster_old["name"] == cluster:
            msg = f"Old and new clusters can't have the same name: {cluster}"
            raise ValueError(msg)

        # Replacement pattern for clusters.
        cluster_str_old = f"IN CLUSTER {sql.identifier(cluster_old['name'], True)}"