Implementation:Astronomer Astronomer cosmos Dbt Project Utils

Knowledge Sources	Astronomer_Astronomer_cosmos
Domains	Filesystem, dbt_Project
Last Updated	2026-02-07 17:00 GMT

Overview

The dbt project utilities module provides filesystem and environment management functions for preparing, copying, and symlinking dbt project artifacts during Cosmos task execution.

Description

This 181-line module contains a collection of utility functions that handle the mechanical aspects of dbt project setup within an Airflow worker environment.

has_non_empty_dependencies_file checks whether a dbt project contains a non-empty packages.yml or dependencies.yml file, which determines whether dbt deps needs to run before execution.

get_dbt_packages_subpath resolves the path to the dbt_packages directory relative to the project root, accounting for custom configurations in dbt_project.yml.

copy_dbt_packages copies the dbt_packages directory from a source location to a target project directory, ensuring that pre-installed packages are available without re-downloading.

copy_manifest_file_if_exists conditionally copies the manifest.json file into the target directory, enabling partial parsing and cross-referencing without a full project compile.

create_symlinks establishes symbolic links from a temporary working directory back to the original project files, allowing Cosmos to isolate task execution without duplicating the entire project tree. This is critical for concurrent task execution on the same worker.

get_partial_parse_path locates the partial_parse.msgpack file, which dbt uses to speed up re-compilation by caching parse results.

environ is a context manager that temporarily sets environment variables for the duration of a block and restores the original values on exit. This is used to inject dbt-specific environment variables (e.g., DBT_PROFILES_DIR) without polluting the global process environment.

change_working_directory is a context manager that temporarily changes the current working directory and restores it on exit, ensuring dbt commands execute in the correct project root.

Usage

Use these utilities when building custom execution strategies or operators that need to prepare a dbt project directory before running dbt commands. They are called internally by Cosmos's local execution operators but are available for direct use in custom implementations or testing scenarios.

Code Reference

Source Location

• Repository: Astronomer_Astronomer_cosmos
• File: cosmos/dbt/project.py
• Lines: Full module (181 lines)

Signature

def has_non_empty_dependencies_file(project_path: Path) -> bool:
    ...

def get_dbt_packages_subpath(project_path: Path) -> Path:
    ...

def copy_dbt_packages(source: Path, destination: Path) -> None:
    ...

def copy_manifest_file_if_exists(source: Path, destination: Path) -> None:
    ...

def create_symlinks(project_path: Path, tmp_dir: Path, ignore_dbt_packages: bool = False) -> None:
    ...

def get_partial_parse_path(project_path: Path) -> Path:
    ...

@contextmanager
def environ(env_vars: dict[str, str]) -> Generator[None, None, None]:
    ...

@contextmanager
def change_working_directory(path: Path) -> Generator[None, None, None]:
    ...

Import

from cosmos.dbt.project import create_symlinks
from cosmos.dbt.project import copy_dbt_packages
from cosmos.dbt.project import environ
from cosmos.dbt.project import change_working_directory
from cosmos.dbt.project import has_non_empty_dependencies_file
from cosmos.dbt.project import copy_manifest_file_if_exists

I/O Contract

Inputs

Name	Type	Required	Description
project_path	Path	Yes	Absolute path to the root of the dbt project directory
source	Path	Yes	Source path from which to copy dbt_packages or manifest files
destination	Path	Yes	Target path where files or directories will be copied to
tmp_dir	Path	Yes	Temporary working directory where symlinks will be created
ignore_dbt_packages	bool	No	When True, skip symlinking the dbt_packages directory (defaults to False)
env_vars	dict[str, str]	Yes	Dictionary of environment variable names and values to set temporarily
path	Path	Yes	Directory to change the working directory to within the context manager

Outputs

Name	Type	Description
has_dependencies	bool	(has_non_empty_dependencies_file) Whether the project has a non-empty packages or dependencies file
packages_subpath	Path	(get_dbt_packages_subpath) Resolved path to the dbt_packages directory
partial_parse_path	Path	(get_partial_parse_path) Path to the partial_parse.msgpack file
(side effects)	None	copy_dbt_packages, copy_manifest_file_if_exists, and create_symlinks perform filesystem operations

Usage Examples

from pathlib import Path
from cosmos.dbt.project import create_symlinks, copy_dbt_packages

project = Path("/opt/airflow/dags/dbt/my_project")
tmp = Path("/tmp/cosmos_work_dir")
tmp.mkdir(exist_ok=True)

# Create symlinks so the tmp directory mirrors the project structure
create_symlinks(project_path=project, tmp_dir=tmp)

# Copy pre-installed packages into the working directory
copy_dbt_packages(source=project, destination=tmp)

from cosmos.dbt.project import environ, change_working_directory
from pathlib import Path

with environ({"DBT_PROFILES_DIR": "/opt/airflow/dags/dbt/profiles"}):
    with change_working_directory(Path("/tmp/cosmos_work_dir")):
        # dbt commands executed here will use the temporary directory
        # and the custom profiles directory
        pass

from pathlib import Path
from cosmos.dbt.project import has_non_empty_dependencies_file

project = Path("/opt/airflow/dags/dbt/my_project")
if has_non_empty_dependencies_file(project):
    print("Need to run dbt deps before execution")

Related Pages

• Environment:Astronomer_Astronomer_cosmos_Python_Airflow_Runtime
• Astronomer_Astronomer_cosmos_DbtRunner_Wrapper -- uses project utilities to prepare the working directory before running dbt commands
• Astronomer_Astronomer_cosmos_FullOutputSubprocessHook -- subprocess execution that relies on correct working directory setup