SnowDDL
Search
⌃K
⚙️

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. 1.
    Create a standard directory with YAML config. You may optionally fill it with YAML files.
  2. 2.
    Create a sub-directory with name __custom (starting with two underscores) in config directory.
  3. 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)