Fix running install_deps=True when using profile_mapping and ExecutionMode.LOCAL

[tatiana]

As of Cosmos 1.2.2, when using profile_mapping, install_deps=True and the ExecutionMode.LOCAL to run dbt commands from within Airflow workers, we are not setting the profile mapping path. ATM, this only works if users use profiles_yml_filepath .

Exception raised:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/cosmos/operators/local.py", line 445, in execute
    self.build_and_run_cmd(context=context, cmd_flags=cmd_flags)
  File "/home/airflow/.local/lib/python3.10/site-packages/cosmos/operators/local.py", line 357, in build_and_run_cmd
    result = self.run_command(cmd=dbt_cmd, env=env, context=context)
  File "/home/airflow/.local/lib/python3.10/site-packages/cosmos/operators/local.py", line 243, in run_command
    self.exception_handling(result)
  File "/home/airflow/.local/lib/python3.10/site-packages/cosmos/operators/local.py", line 124, in exception_handling
    raise AirflowException(
airflow.exceptions.AirflowException: ('dbt command failed. The command returned a non-zero exit code 2. Details: ', '\x1b[0m11:56:26  Running with dbt=1.5.0', '\x1b[0m11:56:29  Encountered an error:', 'Compilation Error', '  dbt found 1 package(s) specified in packages.yml, but only 0 package(s) installed in dbt_packages. Run "dbt deps" to install package dependencies.')

Although we're handling this properly during DAG parsing:

astronomer-cosmos/cosmos/dbt/graph.py

Lines 247 to 266 in 8e2d590

	self.local_flags = [
	"--project-dir",
	str(tmpdir),
	"--profiles-dir",
	str(profile_path.parent),
	"--profile",
	self.profile_config.profile_name,
	"--target",
	self.profile_config.target_name,
	]
	self.log_dir = Path(env.get(DBT_LOG_PATH_ENVVAR) or tmpdir_path / DBT_LOG_DIR_NAME)
	self.target_dir = Path(env.get(DBT_TARGET_PATH_ENVVAR) or tmpdir_path / DBT_TARGET_DIR_NAME)
	env[DBT_LOG_PATH_ENVVAR] = str(self.log_dir)
	env[DBT_TARGET_PATH_ENVVAR] = str(self.target_dir)
	if self.render_config.dbt_deps:
	deps_command = [self.dbt_cmd, "deps"]
	deps_command.extend(self.local_flags)
	stdout = run_command(deps_command, tmpdir_path, env)
	logger.debug("dbt deps output: %s", stdout)

We are indeed not passing the additional parameters in task execution:

astronomer-cosmos/cosmos/operators/local.py

Lines 209 to 216 in 8e2d590

	# if we need to install deps, do so
	if self.install_deps:
	self.run_subprocess(
	command=[self.dbt_executable_path, "deps"],
	env=env,
	output_encoding=self.output_encoding,
	cwd=tmp_project_dir,
	)

The goal of this ticket is to make this logic consistent, fixing the issue in the local.py.

Joppe Vos initially reported this issue in the #airflow-dbt Airflow slack channel:
https://apache-airflow.slack.com/archives/C059CC42E9W/p1699365889141509

[dosubot]

🤖

Hello @tatiana,

I'm Dosu, a friendly helper bot. I'm here to assist you with your technical difficulties, answer any questions you may have, and help you contribute to our repository. While we wait for a human maintainer to get back to you, feel free to ask me anything. Let's work on this together!

Thank you for bringing this issue to our attention. It appears that the problem is related to the environment variables not being set correctly during task execution in the DbtLocalBaseOperator class. This is likely due to the fact that the environment variables are being set in the calculate_openlineage_events_completes method, which is not called during task execution.

To resolve this, you can modify the run_command method in the DbtLocalBaseOperator class to set the environment variables before running the command. Here's how you can do it:

def run_command(
    self,
    cmd: list[str],
    env: dict[str, str | bytes | os.PathLike[Any]],
    context: Context,
) -> FullOutputSubprocessResult:
    """
    Copies the dbt project to a temporary directory and runs the command.
    """
    with tempfile.TemporaryDirectory() as tmp_dir:
        logger.info(
            "Cloning project to writable temp directory %s from %s",
            tmp_dir,
            self.project_dir,
        )

        # need a subfolder because shutil.copytree will fail if the destination dir already exists
        tmp_project_dir = os.path.join(tmp_dir, "dbt_project")
        shutil.copytree(
            self.project_dir,
            tmp_project_dir,
        )

        # Set environment variables
        for key, value in env.items():
            os.environ[key] = str(value)

        subprocess_result: FullOutputSubprocessResult = self.subprocess_hook.run_command(cmd, env)
        return subprocess_result

This modification ensures that the environment variables are set before the command is run, which should resolve the issue of the profile mapping path not being set during task execution.

As for the install_deps=True parameter, I wasn't able to find specific information about its role within the Cosmos codebase. It might be beneficial to review the documentation or source code where this parameter is used for more context.

Please let me know if this solution works for you or if you need further assistance.

Sources

• cosmos/operators/local.py

---

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.