Function defs

misc/python/materialize/mzexplore/extract.py:39–118

Extract CREATE statements for selected catalog items. Processes only items that match the ILIKE pattern defined by the parameter triple (DATABASE, SCHEMA, NAME).

(
    target: Path,
    database: str,
    schema: str,
    name: str,
    db_port: int,
    db_host: str,
    db_user: str,
    db_pass: str | None,
    db_require_ssl: bool,
    mzfmt: bool,
)

def defs(
    target: Path,
    database: str,
    schema: str,
    name: str,
    db_port: int,
    db_host: str,
    db_user: str,
    db_pass: str | None,
    db_require_ssl: bool,
    mzfmt: bool,
) -> None:
    """
    Extract CREATE statements for selected catalog items.

    Processes only items that match the ILIKE pattern defined by the parameter
    triple (DATABASE, SCHEMA, NAME).
    """

    # Ensure that the target dir exists.
    target.mkdir(parents=True, exist_ok=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("""
                -- id: $id
                -- oid: $oid
                $create_sql
                """).lstrip())

        # Extract materialized view definitions
        # -------------------------------------

        for item in db.catalog_items(database, schema, name, system=False):
            item_database = sql.identifier(item["database"])
            item_schema = sql.identifier(item["schema"])
            item_name = sql.identifier(item["name"])
            fqname = f"{item_database}.{item_schema}.{item_name}"

            try:
                item_type = ItemType(item["type"])
            except ValueError:
                warn(f"Unsupported item type `{item['type']}` for {fqname}")
                continue

            create_file = CreateFile(
                database=item["database"],
                schema=item["schema"],
                name=item["name"],
                item_type=item_type,