trino_spells_mover

trino_spells_mover.py

import os
import re
import shutil

def replace_with_alias_macros(model_contents):
    """
    Replace alias = 'some_alias' with alias('some_alias', legacy_model=True)
    """
    pattern = r"alias\s*=\s*'([^']*)'"
    # replace alias with legacy alias
    transformation_string = "alias = alias('\\1')"
    return re.sub(pattern, transformation_string, model_contents)

# Get current dir
dir_path = os.getcwd()

for root, dirs, files in os.walk(dir_path):

    for file in files:

        # Check for .sql file
        if file.endswith('.sql'):
            # Get full file path
            full_path = os.path.join(root, file)

            with open(full_path) as f:
                model_contents = f.read()

            # replace alias with alias macro
            model_contents = replace_with_alias_macros(model_contents)

            # replace or create tags
            if "tags = ['static']" in model_contents:
                model_contents = model_contents.replace("tags = ['static']", "tags = ['static', 'dunesql']")
            elif "tags=['static']" in model_contents:
                model_contents = model_contents.replace("tags=['static']", "tags=['static', 'dunesql']")
            elif "tags=['prod_exclude']" in model_contents:
                model_contents = model_contents.replace("tags=['prod_exclude']", "tags=['prod_exclude', 'dunesql']")
            else:
                # no existing tags, creating new tags
                model_contents = model_contents.replace("config(", "config(tags=['dunesql'],")

            pattern = r'delta_prod\.(\w+)\.(\w+)'
            model_contents = re.sub(pattern, lambda m: "{{ ref('{}_{}') }}".format(m.group(1), m.group(2)), model_contents)

            # Write only to original .sql file
            with open(full_path, 'w') as f:
                f.write(model_contents)