trying to use subprocess python module to apply some commands if a dbt macro logic is not met
macro:
{% macro check_select_arg(run_all=False, dbt_command='dbt run') %}
{% do log("ON START ✅", info=True) %}
{% set subprocess = imprt( 'subprocess' ) %}
{% if run_all %}
{{subprocess.run(dbt_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)}}
{% elif not invocation_args_dict.select
and target.name not in ['prod']
and invocation_args_dict.which in ['build', 'run', 'test', 'source', 'snapshot', 'seed'] %}
{{ exceptions.raise_compiler_error("❌ Error: You must provide at least one select argument") }}
{% endif %}
{% endmacro %}
my python code:
import jinja2
import importlib
FOLDER_PATH = "check_select_arg.sql"
environment = jinja2.Environment()
template_file = open(FOLDER_PATH, "r")
template_string = template_file.read()
template_file.close()
template = environment.from_string(template_string)
template.render(imprt = importlib.import_module)
the problem is that when i run the command dbt run-operation check_select_arg --arg "{run_all: True, dbt_command: 'dbt run'}"
it returns the following error:
Encountered an error while running operation: Compilation Error 'imprt' is undefined
i was expecting to be able to access subprocess and thus, run the dbt command.
thanks all for the help in advance