⚙ïļProgrammatic config

It is possible to extend and modify SnowDDL config programmatically using pure Python.

A few examples of real business use cases which can be implemented with this technique:

  • Get list of users dynamically from single sign-on data provider;

  • Generate a view for each table in specific schemas;

  • Generate masking policy for each table containing columns named "email" and "phone";

  • Skip certain types of objects in DEV environment;

There are not restrictions. Any external data source and any Python package can be used.

Implementation steps

  1. Create a standard directory with YAML config. You may optionally fill it with YAML files.

  2. Create a sub-directory with name __custom (starting with two underscores) in config directory.

  3. Place one or more python modules (.py files) in __custom sub-directory.

During SnowDDL execution YAML configs are resolved first. After that Python modules are resolved one-by-one in alphanumeric order.

It is highly recommended to start module names with zero-padded numbers to make sure you have a precise control of resolution order, for example: 01_foo.py, 02_bar.py, 03_baz.py.

Module requirements

  • Each module should have a function with name handler, which accepts instance of SnowDDLConfig as a single argument. This function does not return anything.

  • In handler function you may build blueprint objects representing the desired state of objects in Snowflake, and use config methods .add_blueprint() and .remove_blueprint() to manipulate the collection of blueprints.

  • You may access existing blueprints using methods .get_blueprints_by_type() and .get_blueprints_by_type_and_pattern().

Examples

from snowddl import DataType, Ident, TableBlueprint, TableColumn, SchemaObjectIdent, SnowDDLConfig


def handler(config: SnowDDLConfig):
    # Add custom tables
    for i in range(1, 5):
        bp = TableBlueprint(
            full_name=SchemaObjectIdent(config.env_prefix, "test_db", "test_schema", f"custom_table_{i}"),
            columns=[
                TableColumn(
                    name=Ident("id"),
                    type=DataType("NUMBER(38,0)"),
                ),
                TableColumn(
                    name=Ident("name"),
                    type=DataType("VARCHAR(255)"),
                ),
            ],
            is_transient=True,
            comment="This table was created programmatically",
        )

        config.add_blueprint(bp)
  • Example of Python module which scans current config for custom tables and generates a consolidated view dynamically:

from snowddl import SchemaObjectIdent, SnowDDLConfig, TableBlueprint, ViewBlueprint


def handler(config: SnowDDLConfig):
    # Add view combining all custom tables
    parts = []

    for full_name, bp in config.get_blueprints_by_type_and_pattern(TableBlueprint, "test_db.test_schema.custom_table_*").items():
        parts.append(f"SELECT id, name FROM {full_name}")

    bp = ViewBlueprint(
        full_name=SchemaObjectIdent(config.env_prefix, "test_db", "test_schema", "custom_view"),
        text="\nUNION ALL\n".join(parts),
        comment="This view was created programmatically",
    )

    config.add_blueprint(bp)

Last updated