Modules Reference

Auto-generated API documentation for brix modules.

Profile Service

brix.modules.dbt.profile.service

Profile management service for dbt profiles.yml.

Handles loading templates, validating profiles, and writing to disk.

ProfileConfig

Bases: BaseSettings

Profile configuration from environment variables.

Environment variables

BRIX_DBT_PROFILE_PATH: Override default profile path

Source code in src/brix/modules/dbt/profile/service.py

class ProfileConfig(BaseSettings):
    """Profile configuration from environment variables.

    Environment variables:
        BRIX_DBT_PROFILE_PATH: Override default profile path
    """

    model_config = SettingsConfigDict(
        env_prefix="BRIX_DBT_",
        case_sensitive=False,
    )

    profile_path: Path | None = None

ProfileExistsError

Bases: Exception

Raised when profile already exists and force is not set.

Source code in src/brix/modules/dbt/profile/service.py

class ProfileExistsError(Exception):
    """Raised when profile already exists and force is not set."""

ProfileInitResult

Result of profile initialization.

Source code in src/brix/modules/dbt/profile/service.py

class ProfileInitResult:
    """Result of profile initialization."""

    def __init__(
        self,
        *,
        success: bool,
        path: Path,
        action: Literal["created", "overwritten", "skipped"],
        message: str,
    ) -> None:
        """Initialize profile init result.

        Args:
            success: Whether initialization succeeded
            path: Path to the profile file
            action: What action was taken (created, overwritten, skipped)
            message: Human-readable result message
        """
        self.success = success
        self.path = path
        self.action = action
        self.message = message

__init__(*, success, path, action, message)

Initialize profile init result.

Parameters:

Name	Type	Description	Default
success	bool	Whether initialization succeeded	required
path	Path	Path to the profile file	required
action	Literal['created', 'overwritten', 'skipped']	What action was taken (created, overwritten, skipped)	required
message	str	Human-readable result message	required

Source code in src/brix/modules/dbt/profile/service.py

def __init__(
    self,
    *,
    success: bool,
    path: Path,
    action: Literal["created", "overwritten", "skipped"],
    message: str,
) -> None:
    """Initialize profile init result.

    Args:
        success: Whether initialization succeeded
        path: Path to the profile file
        action: What action was taken (created, overwritten, skipped)
        message: Human-readable result message
    """
    self.success = success
    self.path = path
    self.action = action
    self.message = message

get_default_profile_path()

Get the default profile path, checking env var first.

Returns:

Type	Description
Path	Path from BRIX_DBT_PROFILE_PATH env var, or ~/.dbt/profiles.yml

Source code in src/brix/modules/dbt/profile/service.py

def get_default_profile_path() -> Path:
    """Get the default profile path, checking env var first.

    Returns:
        Path from BRIX_DBT_PROFILE_PATH env var, or ~/.dbt/profiles.yml
    """
    config = ProfileConfig()
    return config.profile_path or DEFAULT_PROFILE_PATH

load_template(template_name='profiles.yml')

Load and validate the bundled profile template.

Parameters:

Name	Type	Description	Default
template_name	str	Name of the template file	'profiles.yml'

Returns:

Type	Description
tuple[str, DbtProfiles]	Tuple of (raw template content, validated DbtProfiles)

Raises:

Type	Description
FileNotFoundError	If template doesn't exist
ValueError	If template is invalid YAML or doesn't match schema

Source code in src/brix/modules/dbt/profile/service.py

def load_template(template_name: str = "profiles.yml") -> tuple[str, DbtProfiles]:
    """Load and validate the bundled profile template.

    Args:
        template_name: Name of the template file

    Returns:
        Tuple of (raw template content, validated DbtProfiles)

    Raises:
        FileNotFoundError: If template doesn't exist
        ValueError: If template is invalid YAML or doesn't match schema
    """
    logger = get_logger()
    logger.debug("Loading template: %s", template_name)

    content = get_template(template_name)
    profiles = DbtProfiles.from_yaml(content)

    logger.debug("Template validated successfully")
    return content, profiles

init_profile(profile_path=None, *, force=False, template_name='profiles.yml')

Initialize a dbt profile from template.

Parameters:

Name	Type	Description	Default
profile_path	Path | None	Target path for profiles.yml (uses default if None)	None
force	bool	Overwrite existing file if True	False
template_name	str	Name of template to use	'profiles.yml'

Returns:

Type	Description
ProfileInitResult	ProfileInitResult with success status and details

Raises:

Type	Description
ProfileExistsError	If file exists and force is False
FileNotFoundError	If template doesn't exist
ValueError	If template validation fails

Source code in src/brix/modules/dbt/profile/service.py

def init_profile(
    profile_path: Path | None = None,
    *,
    force: bool = False,
    template_name: str = "profiles.yml",
) -> ProfileInitResult:
    """Initialize a dbt profile from template.

    Args:
        profile_path: Target path for profiles.yml (uses default if None)
        force: Overwrite existing file if True
        template_name: Name of template to use

    Returns:
        ProfileInitResult with success status and details

    Raises:
        ProfileExistsError: If file exists and force is False
        FileNotFoundError: If template doesn't exist
        ValueError: If template validation fails
    """
    logger = get_logger()

    # Determine target path
    target_path = profile_path or get_default_profile_path()
    logger.debug("Target profile path: %s", target_path)

    # Check if file exists
    if target_path.exists() and not force:
        msg = f"Profile already exists at {target_path}. Use --force to overwrite."
        raise ProfileExistsError(msg)

    # Load and validate template
    content, _ = load_template(template_name)

    # Ensure parent directory exists
    target_path.parent.mkdir(parents=True, exist_ok=True)

    # Determine action for result
    action: Literal["created", "overwritten", "skipped"] = "overwritten" if target_path.exists() else "created"

    # Write profile
    target_path.write_text(content)
    logger.info("Profile %s at %s", action, target_path)

    return ProfileInitResult(
        success=True,
        path=target_path,
        action=action,
        message=f"Profile {action} at {target_path}",
    )

Profile Editor

brix.modules.dbt.profile.editor

Profile editing service for dbt profiles.yml.

Provides CRUD operations for profiles and outputs with atomic save-on-change behavior.

ProfileNotFoundError

Bases: Exception

Raised when a profile does not exist.

Source code in src/brix/modules/dbt/profile/editor.py

class ProfileNotFoundError(Exception):
    """Raised when a profile does not exist."""

OutputNotFoundError

Bases: Exception

Raised when an output does not exist.

Source code in src/brix/modules/dbt/profile/editor.py

class OutputNotFoundError(Exception):
    """Raised when an output does not exist."""

ProfileAlreadyExistsError

Bases: Exception

Raised when attempting to create a profile that already exists.

Source code in src/brix/modules/dbt/profile/editor.py

class ProfileAlreadyExistsError(Exception):
    """Raised when attempting to create a profile that already exists."""

OutputAlreadyExistsError

Bases: Exception

Raised when attempting to create an output that already exists.

Source code in src/brix/modules/dbt/profile/editor.py

class OutputAlreadyExistsError(Exception):
    """Raised when attempting to create an output that already exists."""

load_profiles(path=None)

Load profiles from disk.

Parameters:

Name	Type	Description	Default
path	Path | None	Path to profiles.yml, uses default if None	None

Returns:

Type	Description
DbtProfiles	Parsed DbtProfiles instance

Raises:

Type	Description
FileNotFoundError	If file doesn't exist
ValueError	If YAML is invalid

Source code in src/brix/modules/dbt/profile/editor.py

def load_profiles(path: Path | None = None) -> DbtProfiles:
    """Load profiles from disk.

    Args:
        path: Path to profiles.yml, uses default if None

    Returns:
        Parsed DbtProfiles instance

    Raises:
        FileNotFoundError: If file doesn't exist
        ValueError: If YAML is invalid
    """
    target_path = path or get_default_profile_path()
    logger = get_logger()
    logger.debug("Loading profiles from %s", target_path)
    return DbtProfiles.from_file(target_path)

save_profiles(profiles, path=None)

Validate and save profiles to disk.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance to save	required
path	Path | None	Path to profiles.yml, uses default if None	None

Raises:

Type	Description
ValueError	If profiles fail validation
IOError	If file cannot be written

Source code in src/brix/modules/dbt/profile/editor.py

def save_profiles(profiles: DbtProfiles, path: Path | None = None) -> None:
    """Validate and save profiles to disk.

    Args:
        profiles: DbtProfiles instance to save
        path: Path to profiles.yml, uses default if None

    Raises:
        ValueError: If profiles fail validation
        IOError: If file cannot be written
    """
    target_path = path or get_default_profile_path()
    logger = get_logger()

    # Validate by re-parsing (ensures YAML roundtrip is valid)
    yaml_content = profiles.to_yaml()
    DbtProfiles.from_yaml(yaml_content)

    # Ensure parent directory exists
    target_path.parent.mkdir(parents=True, exist_ok=True)

    # Write to disk
    target_path.write_text(yaml_content)
    logger.debug("Saved profiles to %s", target_path)

get_profile_names(profiles)

Get list of profile names.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required

Returns:

Type	Description
list[str]	List of profile names

Source code in src/brix/modules/dbt/profile/editor.py

def get_profile_names(profiles: DbtProfiles) -> list[str]:
    """Get list of profile names.

    Args:
        profiles: DbtProfiles instance

    Returns:
        List of profile names
    """
    return list(profiles.root.keys())

get_output_names(profiles, profile_name)

Get list of output names for a profile.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required

Returns:

Type	Description
list[str]	List of output names

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist

Source code in src/brix/modules/dbt/profile/editor.py

def get_output_names(profiles: DbtProfiles, profile_name: str) -> list[str]:
    """Get list of output names for a profile.

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile

    Returns:
        List of output names

    Raises:
        ProfileNotFoundError: If profile doesn't exist
    """
    if profile_name not in profiles.root:
        msg = f"Profile '{profile_name}' not found"
        raise ProfileNotFoundError(msg)
    return list(profiles.root[profile_name].outputs.keys())

add_profile(profiles, name, target, output_name, output_config)

Add a new profile.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
name	str	Profile name	required
target	str	Default target name	required
output_name	str	Initial output name	required
output_config	OutputConfig	Initial output configuration (DuckDbOutput or DatabricksOutput)	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileAlreadyExistsError	If profile already exists

Source code in src/brix/modules/dbt/profile/editor.py

def add_profile(
    profiles: DbtProfiles,
    name: str,
    target: str,
    output_name: str,
    output_config: OutputConfig,
) -> DbtProfiles:
    """Add a new profile.

    Args:
        profiles: DbtProfiles instance
        name: Profile name
        target: Default target name
        output_name: Initial output name
        output_config: Initial output configuration (DuckDbOutput or DatabricksOutput)

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileAlreadyExistsError: If profile already exists
    """
    if name in profiles.root:
        msg = f"Profile '{name}' already exists"
        raise ProfileAlreadyExistsError(msg)

    profiles.root[name] = ProfileTarget(
        target=target,
        outputs={output_name: output_config},
    )
    return profiles

update_profile_target(profiles, name, target)

Update a profile's default target.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
name	str	Profile name	required
target	str	New default target name	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist

Source code in src/brix/modules/dbt/profile/editor.py

def update_profile_target(profiles: DbtProfiles, name: str, target: str) -> DbtProfiles:
    """Update a profile's default target.

    Args:
        profiles: DbtProfiles instance
        name: Profile name
        target: New default target name

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
    """
    if name not in profiles.root:
        msg = f"Profile '{name}' not found"
        raise ProfileNotFoundError(msg)

    profiles.root[name].target = target
    return profiles

delete_profile(profiles, name)

Delete a profile.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
name	str	Profile name to delete	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist

Source code in src/brix/modules/dbt/profile/editor.py

def delete_profile(profiles: DbtProfiles, name: str) -> DbtProfiles:
    """Delete a profile.

    Args:
        profiles: DbtProfiles instance
        name: Profile name to delete

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
    """
    if name not in profiles.root:
        msg = f"Profile '{name}' not found"
        raise ProfileNotFoundError(msg)

    del profiles.root[name]
    return profiles

add_output(profiles, profile_name, output_name, output_config)

Add an output to a profile.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required
output_name	str	Name for the new output	required
output_config	OutputConfig	Output configuration (DuckDbOutput or DatabricksOutput)	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist
OutputAlreadyExistsError	If output already exists

Source code in src/brix/modules/dbt/profile/editor.py

def add_output(
    profiles: DbtProfiles,
    profile_name: str,
    output_name: str,
    output_config: OutputConfig,
) -> DbtProfiles:
    """Add an output to a profile.

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile
        output_name: Name for the new output
        output_config: Output configuration (DuckDbOutput or DatabricksOutput)

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
        OutputAlreadyExistsError: If output already exists
    """
    if profile_name not in profiles.root:
        msg = f"Profile '{profile_name}' not found"
        raise ProfileNotFoundError(msg)

    if output_name in profiles.root[profile_name].outputs:
        msg = f"Output '{output_name}' already exists in profile '{profile_name}'"
        raise OutputAlreadyExistsError(msg)

    profiles.root[profile_name].outputs[output_name] = output_config
    return profiles

update_output(profiles, profile_name, output_name, *, path=None, threads=None)

Update a DuckDB output's configuration (legacy interface).

For updating any adapter type, use update_output_fields() instead.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required
output_name	str	Name of the output	required
path	str | None	New path value (optional, DuckDB only)	None
threads	int | None	New threads value (optional)	None

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist
OutputNotFoundError	If output doesn't exist
ValueError	If threads is not a positive integer

Source code in src/brix/modules/dbt/profile/editor.py

def update_output(
    profiles: DbtProfiles,
    profile_name: str,
    output_name: str,
    *,
    path: str | None = None,
    threads: int | None = None,
) -> DbtProfiles:
    """Update a DuckDB output's configuration (legacy interface).

    For updating any adapter type, use update_output_fields() instead.

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile
        output_name: Name of the output
        path: New path value (optional, DuckDB only)
        threads: New threads value (optional)

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
        OutputNotFoundError: If output doesn't exist
        ValueError: If threads is not a positive integer
    """
    updates: dict[str, Any] = {}
    if path is not None:
        updates["path"] = path
    if threads is not None:
        updates["threads"] = threads
    return update_output_fields(profiles, profile_name, output_name, updates)

update_output_fields(profiles, profile_name, output_name, updates)

Update an output's configuration with arbitrary field updates.

Works with any adapter type (DuckDB, Databricks, etc.).

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required
output_name	str	Name of the output	required
updates	dict[str, Any]	Dictionary of field names to new values	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist
OutputNotFoundError	If output doesn't exist
ValueError	If validation fails (e.g., threads < 1)

Source code in src/brix/modules/dbt/profile/editor.py

def update_output_fields(
    profiles: DbtProfiles,
    profile_name: str,
    output_name: str,
    updates: dict[str, Any],
) -> DbtProfiles:
    """Update an output's configuration with arbitrary field updates.

    Works with any adapter type (DuckDB, Databricks, etc.).

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile
        output_name: Name of the output
        updates: Dictionary of field names to new values

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
        OutputNotFoundError: If output doesn't exist
        ValueError: If validation fails (e.g., threads < 1)
    """
    if profile_name not in profiles.root:
        msg = f"Profile '{profile_name}' not found"
        raise ProfileNotFoundError(msg)

    if output_name not in profiles.root[profile_name].outputs:
        msg = f"Output '{output_name}' not found in profile '{profile_name}'"
        raise OutputNotFoundError(msg)

    output = profiles.root[profile_name].outputs[output_name]

    # Validate threads if being updated
    if "threads" in updates:
        threads = updates["threads"]
        if threads is not None and threads < 1:
            msg = "threads must be a positive integer"
            raise ValueError(msg)

    # Apply updates to the output
    for field, value in updates.items():
        if value is not None:
            setattr(output, field, value)

    return profiles

delete_output(profiles, profile_name, output_name)

Delete an output from a profile.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required
output_name	str	Name of the output to delete	required

Returns:

Type	Description
DbtProfiles	Updated DbtProfiles instance

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist
OutputNotFoundError	If output doesn't exist
ValueError	If this is the last output in the profile

Source code in src/brix/modules/dbt/profile/editor.py

def delete_output(
    profiles: DbtProfiles,
    profile_name: str,
    output_name: str,
) -> DbtProfiles:
    """Delete an output from a profile.

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile
        output_name: Name of the output to delete

    Returns:
        Updated DbtProfiles instance

    Raises:
        ProfileNotFoundError: If profile doesn't exist
        OutputNotFoundError: If output doesn't exist
        ValueError: If this is the last output in the profile
    """
    if profile_name not in profiles.root:
        msg = f"Profile '{profile_name}' not found"
        raise ProfileNotFoundError(msg)

    if output_name not in profiles.root[profile_name].outputs:
        msg = f"Output '{output_name}' not found in profile '{profile_name}'"
        raise OutputNotFoundError(msg)

    if len(profiles.root[profile_name].outputs) == 1:
        msg = f"Cannot delete last output from profile '{profile_name}'. Delete the profile instead."
        raise ValueError(msg)

    del profiles.root[profile_name].outputs[output_name]
    return profiles

get_output(profiles, profile_name, output_name)

Get an output configuration.

Parameters:

Name	Type	Description	Default
profiles	DbtProfiles	DbtProfiles instance	required
profile_name	str	Name of the profile	required
output_name	str	Name of the output	required

Returns:

Type	Description
OutputConfig	Output configuration (DuckDbOutput or DatabricksOutput)

Raises:

Type	Description
ProfileNotFoundError	If profile doesn't exist
OutputNotFoundError	If output doesn't exist

Source code in src/brix/modules/dbt/profile/editor.py

def get_output(profiles: DbtProfiles, profile_name: str, output_name: str) -> OutputConfig:
    """Get an output configuration.

    Args:
        profiles: DbtProfiles instance
        profile_name: Name of the profile
        output_name: Name of the output

    Returns:
        Output configuration (DuckDbOutput or DatabricksOutput)

    Raises:
        ProfileNotFoundError: If profile doesn't exist
        OutputNotFoundError: If output doesn't exist
    """
    if profile_name not in profiles.root:
        msg = f"Profile '{profile_name}' not found"
        raise ProfileNotFoundError(msg)

    if output_name not in profiles.root[profile_name].outputs:
        msg = f"Output '{output_name}' not found in profile '{profile_name}'"
        raise OutputNotFoundError(msg)

    return profiles.root[profile_name].outputs[output_name]

Project Service

brix.modules.dbt.project.service

Project management service for dbt projects.

Handles project initialization, path resolution, and package version fetching.

ProjectConfig

Bases: BaseSettings

Project configuration from environment variables.

Environment variables

BRIX_DBT_PROJECT_BASE_DIR: Default base directory for projects

Source code in src/brix/modules/dbt/project/service.py

class ProjectConfig(BaseSettings):
    """Project configuration from environment variables.

    Environment variables:
        BRIX_DBT_PROJECT_BASE_DIR: Default base directory for projects
    """

    model_config = SettingsConfigDict(
        env_prefix="BRIX_DBT_PROJECT_",
        case_sensitive=False,
    )

    base_dir: Path | None = None

ProjectExistsError

Bases: Exception

Raised when project already exists and force is not set.

Source code in src/brix/modules/dbt/project/service.py

class ProjectExistsError(Exception):
    """Raised when project already exists and force is not set."""

ProjectInitResult dataclass

Result of project initialization.

Source code in src/brix/modules/dbt/project/service.py

@dataclass
class ProjectInitResult:
    """Result of project initialization."""

    success: bool
    project_path: Path
    action: Literal["created", "overwritten", "skipped"]
    message: str
    files_created: list[str] = field(default_factory=list)

resolve_project_path(project_name, base_dir=None, team=None)

Resolve the final project path from components.

Parameters:

Name	Type	Description	Default
project_name	str	Name of the project (becomes directory name)	required
base_dir	Path | None	Base directory (uses env var or cwd if None)	None
team	str | None	Optional team subdirectory	None

Returns:

Type	Description
Path	Resolved absolute path to project directory

Example

resolve_project_path("my_project") PosixPath('/current/dir/my_project') resolve_project_path("my_project", Path("assets/dbt_projects"), "analytics") PosixPath('/current/dir/assets/dbt_projects/analytics/my_project')

Source code in src/brix/modules/dbt/project/service.py

def resolve_project_path(
    project_name: str,
    base_dir: Path | None = None,
    team: str | None = None,
) -> Path:
    """Resolve the final project path from components.

    Args:
        project_name: Name of the project (becomes directory name)
        base_dir: Base directory (uses env var or cwd if None)
        team: Optional team subdirectory

    Returns:
        Resolved absolute path to project directory

    Example:
        >>> resolve_project_path("my_project")
        PosixPath('/current/dir/my_project')
        >>> resolve_project_path("my_project", Path("assets/dbt_projects"), "analytics")
        PosixPath('/current/dir/assets/dbt_projects/analytics/my_project')
    """
    config = ProjectConfig()
    effective_base = base_dir or config.base_dir or Path.cwd()

    # Make path absolute if relative
    if not effective_base.is_absolute():
        effective_base = Path.cwd() / effective_base

    if team:
        return effective_base / team / project_name
    return effective_base / project_name

fetch_package_version(package)

Fetch the latest version of a package from dbt Hub.

Parameters:

Name	Type	Description	Default
package	str	Package name (e.g., "dbt-labs/dbt_utils")	required

Returns:

Type	Description
str | None	Version string (e.g., ">=1.3.0") or None if fetch fails

Source code in src/brix/modules/dbt/project/service.py

def fetch_package_version(package: str) -> str | None:
    """Fetch the latest version of a package from dbt Hub.

    Args:
        package: Package name (e.g., "dbt-labs/dbt_utils")

    Returns:
        Version string (e.g., ">=1.3.0") or None if fetch fails
    """
    logger = get_logger()

    try:
        namespace, name = package.split("/")
        url = f"https://hub.getdbt.com/api/v1/{namespace}/{name}/latest.json"

        logger.debug("Fetching package version from: %s", url)

        # Simple HTTP GET with timeout
        with urllib.request.urlopen(url, timeout=5) as response:  # noqa: S310
            import json

            data = json.loads(response.read().decode())
            version = data.get("version")
            if version:
                logger.debug("Found version %s for %s", version, package)
                return f">={version}"
    except Exception as e:
        logger.debug("Failed to fetch version for %s: %s", package, e)

    return None

get_package_version(package)

Get the version for a package, with fallback to defaults.

Parameters:

Name	Type	Description	Default
package	str	Package name (e.g., "dbt-labs/dbt_utils")	required

Returns:

Type	Description
str	Version string (e.g., ">=1.0.0")

Source code in src/brix/modules/dbt/project/service.py

def get_package_version(package: str) -> str:
    """Get the version for a package, with fallback to defaults.

    Args:
        package: Package name (e.g., "dbt-labs/dbt_utils")

    Returns:
        Version string (e.g., ">=1.0.0")
    """
    # Try to fetch from API
    version = fetch_package_version(package)
    if version:
        return version

    # Fall back to defaults
    return DEFAULT_PACKAGE_VERSIONS.get(package, ">=0.1.0")

fetch_package_versions_parallel(pkg_names, max_workers=5)

Fetch multiple package versions in parallel.

Parameters:

Name	Type	Description	Default
pkg_names	list[str]	List of package names (e.g., ["dbt-labs/dbt_utils", "elementary-data/elementary"])	required
max_workers	int	Maximum number of concurrent threads	5

Returns:

Type	Description
dict[str, str]	Dictionary mapping package names to version strings

Source code in src/brix/modules/dbt/project/service.py

def fetch_package_versions_parallel(pkg_names: list[str], max_workers: int = 5) -> dict[str, str]:
    """Fetch multiple package versions in parallel.

    Args:
        pkg_names: List of package names (e.g., ["dbt-labs/dbt_utils", "elementary-data/elementary"])
        max_workers: Maximum number of concurrent threads

    Returns:
        Dictionary mapping package names to version strings
    """
    from concurrent.futures import ThreadPoolExecutor, as_completed

    logger = get_logger()
    results: dict[str, str] = {}

    logger.debug("Fetching %d package versions in parallel", len(pkg_names))

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(get_package_version, pkg): pkg for pkg in pkg_names}
        for future in as_completed(futures):
            pkg_name = futures[future]
            try:
                results[pkg_name] = future.result()
            except Exception as e:
                logger.debug("Failed to fetch version for %s: %s", pkg_name, e)
                results[pkg_name] = DEFAULT_PACKAGE_VERSIONS.get(pkg_name, ">=0.1.0")

    return results

create_project_structure(project_path, project_name, profile_name, *, packages=None, materialization=None, persist_docs=False, with_example=False)

Create the dbt project directory structure and files.

Parameters:

Name	Type	Description	Default
project_path	Path	Path to create project in	required
project_name	str	Name of the project	required
profile_name	str	Name of the profile to use	required
packages	list[HubPackage] | None	List of packages to include (uses template default if None)	None
materialization	str | None	Default materialization (view, table, ephemeral)	None
persist_docs	bool	Whether to enable persist_docs for Databricks	False
with_example	bool	Whether to create example model	False

Returns:

Type	Description
list[str]	List of created file paths (relative to project_path)

Source code in src/brix/modules/dbt/project/service.py

def create_project_structure(
    project_path: Path,
    project_name: str,
    profile_name: str,
    *,
    packages: list[HubPackage] | None = None,
    materialization: str | None = None,
    persist_docs: bool = False,
    with_example: bool = False,
) -> list[str]:
    """Create the dbt project directory structure and files.

    Args:
        project_path: Path to create project in
        project_name: Name of the project
        profile_name: Name of the profile to use
        packages: List of packages to include (uses template default if None)
        materialization: Default materialization (view, table, ephemeral)
        persist_docs: Whether to enable persist_docs for Databricks
        with_example: Whether to create example model

    Returns:
        List of created file paths (relative to project_path)
    """
    logger = get_logger()
    created_files: list[str] = []

    # Create main directories
    directories = ["models", "seeds", "tests", "macros", "snapshots", "analyses"]
    for dir_name in directories:
        dir_path = project_path / dir_name
        dir_path.mkdir(parents=True, exist_ok=True)
        # Add .gitkeep to empty directories
        gitkeep = dir_path / ".gitkeep"
        gitkeep.touch()
        created_files.append(f"{dir_name}/.gitkeep")
        logger.debug("Created directory: %s", dir_path)

    # Build dbt_project.yml content
    project_config: dict = {
        "name": project_name,
        "version": "1.0.0",
        "profile": profile_name,
        "config-version": 2,
        "model-paths": ["models"],
        "analysis-paths": ["analyses"],
        "test-paths": ["tests"],
        "seed-paths": ["seeds"],
        "macro-paths": ["macros"],
        "snapshot-paths": ["snapshots"],
        "clean-targets": ["target", "dbt_packages"],
    }

    # Add models config if needed (materialization or persist_docs)
    if materialization or persist_docs:
        models_config: dict = {}
        if materialization and materialization != "view":
            models_config["+materialized"] = materialization
        if persist_docs:
            models_config["+persist_docs"] = {"relation": True, "columns": True}
        if models_config:
            project_config["models"] = {project_name: models_config}

    # Create dbt_project.yml
    project = DbtProject(**project_config)
    project_yml_path = project_path / "dbt_project.yml"
    project_yml_path.write_text(project.to_yaml())
    created_files.append("dbt_project.yml")
    logger.debug("Created: %s", project_yml_path)

    # Create packages.yml only if packages were specified
    if packages is not None:
        dbt_packages = DbtPackages(packages=list(packages))
        packages_content = dbt_packages.to_yaml()
        packages_yml_path = project_path / "packages.yml"
        packages_yml_path.write_text(packages_content)
        created_files.append("packages.yml")
        logger.debug("Created: %s", packages_yml_path)

    # Create .gitignore
    gitignore_content = get_template("dbt_gitignore")
    gitignore_path = project_path / ".gitignore"
    gitignore_path.write_text(gitignore_content)
    created_files.append(".gitignore")
    logger.debug("Created: %s", gitignore_path)

    # Create example model if requested
    if with_example:
        example_dir = project_path / "models" / "example"
        example_dir.mkdir(parents=True, exist_ok=True)

        # Create example model SQL
        model_content = get_template("example_model.sql")
        model_path = example_dir / "my_first_model.sql"
        model_path.write_text(model_content)
        created_files.append("models/example/my_first_model.sql")
        logger.debug("Created: %s", model_path)

        # Create example schema YAML
        schema_content = get_template("example_schema.yml")
        schema_path = example_dir / "schema.yml"
        schema_path.write_text(schema_content)
        created_files.append("models/example/schema.yml")
        logger.debug("Created: %s", schema_path)

    return created_files

init_project(project_name, profile_name, base_dir=None, team=None, *, packages=None, materialization=None, persist_docs=False, with_example=False, force=False)

Initialize a new dbt project.

Parameters:

Name	Type	Description	Default
project_name	str	Name of the project	required
profile_name	str	Name of the profile to use	required
base_dir	Path | None	Base directory for project (uses env var or cwd if None)	None
team	str | None	Optional team subdirectory	None
packages	list[HubPackage] | None	List of packages to include	None
materialization	str | None	Default materialization (view, table, ephemeral)	None
persist_docs	bool	Whether to enable persist_docs for Databricks	False
with_example	bool	Whether to create example model	False
force	bool	Overwrite existing project if True	False

Returns:

Type	Description
ProjectInitResult	ProjectInitResult with success status and details

Raises:

Type	Description
ProjectExistsError	If project exists and force is False
ProjectNameError	If project name is invalid

Source code in src/brix/modules/dbt/project/service.py

def init_project(
    project_name: str,
    profile_name: str,
    base_dir: Path | None = None,
    team: str | None = None,
    *,
    packages: list[HubPackage] | None = None,
    materialization: str | None = None,
    persist_docs: bool = False,
    with_example: bool = False,
    force: bool = False,
) -> ProjectInitResult:
    """Initialize a new dbt project.

    Args:
        project_name: Name of the project
        profile_name: Name of the profile to use
        base_dir: Base directory for project (uses env var or cwd if None)
        team: Optional team subdirectory
        packages: List of packages to include
        materialization: Default materialization (view, table, ephemeral)
        persist_docs: Whether to enable persist_docs for Databricks
        with_example: Whether to create example model
        force: Overwrite existing project if True

    Returns:
        ProjectInitResult with success status and details

    Raises:
        ProjectExistsError: If project exists and force is False
        ProjectNameError: If project name is invalid
    """
    logger = get_logger()

    # Validate project name
    validate_project_name(project_name)

    # Resolve project path
    project_path = resolve_project_path(project_name, base_dir, team)
    logger.debug("Project path: %s", project_path)

    # Check if project exists
    dbt_project_yml = project_path / "dbt_project.yml"
    if dbt_project_yml.exists() and not force:
        msg = f"Project already exists at {project_path}. Use --force to overwrite."
        raise ProjectExistsError(msg)

    # Determine action
    action: Literal["created", "overwritten", "skipped"] = "overwritten" if dbt_project_yml.exists() else "created"

    # Create project directory if needed
    project_path.mkdir(parents=True, exist_ok=True)

    # Create project structure
    files_created = create_project_structure(
        project_path=project_path,
        project_name=project_name,
        profile_name=profile_name,
        packages=packages,
        materialization=materialization,
        persist_docs=persist_docs,
        with_example=with_example,
    )

    logger.info("Project %s at %s", action, project_path)

    return ProjectInitResult(
        success=True,
        project_path=project_path,
        action=action,
        message=f"Project {action} at {project_path}",
        files_created=files_created,
    )

Project Editor

brix.modules.dbt.project.editor

Project editing service for dbt projects.

Provides CRUD operations for dbt_project.yml and packages.yml with atomic save-on-change behavior.

ProjectNotFoundError

Bases: Exception

Raised when dbt_project.yml does not exist.

Source code in src/brix/modules/dbt/project/editor.py

class ProjectNotFoundError(Exception):
    """Raised when dbt_project.yml does not exist."""

PackageNotFoundError

Bases: Exception

Raised when a package does not exist in packages.yml.

Source code in src/brix/modules/dbt/project/editor.py

class PackageNotFoundError(Exception):
    """Raised when a package does not exist in packages.yml."""

PackageAlreadyExistsError

Bases: Exception

Raised when attempting to add a duplicate package.

Source code in src/brix/modules/dbt/project/editor.py

class PackageAlreadyExistsError(Exception):
    """Raised when attempting to add a duplicate package."""

InvalidFieldError

Bases: Exception

Raised when attempting to edit an invalid or restricted field.

Source code in src/brix/modules/dbt/project/editor.py

class InvalidFieldError(Exception):
    """Raised when attempting to edit an invalid or restricted field."""

load_project(path)

Load dbt_project.yml from disk.

Parameters:

Name	Type	Description	Default
path	Path	Path to dbt_project.yml file	required

Returns:

Type	Description
DbtProject	Parsed DbtProject instance

Raises:

Type	Description
ProjectNotFoundError	If file doesn't exist
ValueError	If YAML is invalid

Source code in src/brix/modules/dbt/project/editor.py

def load_project(path: Path) -> DbtProject:
    """Load dbt_project.yml from disk.

    Args:
        path: Path to dbt_project.yml file

    Returns:
        Parsed DbtProject instance

    Raises:
        ProjectNotFoundError: If file doesn't exist
        ValueError: If YAML is invalid
    """
    logger = get_logger()

    if not path.exists():
        msg = f"Project file not found: {path}"
        raise ProjectNotFoundError(msg)

    logger.debug("Loading project from %s", path)
    return DbtProject.from_file(path)

save_project(project, path)

Validate and save dbt_project.yml to disk.

Parameters:

Name	Type	Description	Default
project	DbtProject	DbtProject instance to save	required
path	Path	Path to dbt_project.yml file	required

Raises:

Type	Description
ValueError	If project fails validation
IOError	If file cannot be written

Source code in src/brix/modules/dbt/project/editor.py

def save_project(project: DbtProject, path: Path) -> None:
    """Validate and save dbt_project.yml to disk.

    Args:
        project: DbtProject instance to save
        path: Path to dbt_project.yml file

    Raises:
        ValueError: If project fails validation
        IOError: If file cannot be written
    """
    logger = get_logger()

    # Validate by re-parsing (ensures YAML roundtrip is valid)
    yaml_content = project.to_yaml()
    DbtProject.from_yaml(yaml_content)

    # Ensure parent directory exists
    path.parent.mkdir(parents=True, exist_ok=True)

    # Write to disk
    path.write_text(yaml_content)
    logger.debug("Saved project to %s", path)

load_packages(project_dir)

Load packages.yml from project directory.

Parameters:

Name	Type	Description	Default
project_dir	Path	Path to project directory (or dbt_project.yml file)	required

Returns:

Type	Description
DbtPackages	Parsed DbtPackages instance (empty if file doesn't exist)

Source code in src/brix/modules/dbt/project/editor.py

def load_packages(project_dir: Path) -> DbtPackages:
    """Load packages.yml from project directory.

    Args:
        project_dir: Path to project directory (or dbt_project.yml file)

    Returns:
        Parsed DbtPackages instance (empty if file doesn't exist)
    """
    logger = get_logger()

    # Handle both directory and file paths
    if project_dir.name == "dbt_project.yml":
        project_dir = project_dir.parent

    packages_path = project_dir / "packages.yml"

    if not packages_path.exists():
        logger.debug("No packages.yml found at %s, returning empty", packages_path)
        return DbtPackages(packages=[])

    logger.debug("Loading packages from %s", packages_path)
    return DbtPackages.from_file(packages_path)

save_packages(packages, project_dir)

Save packages.yml to project directory.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance to save	required
project_dir	Path	Path to project directory (or dbt_project.yml file)	required

Raises:

Type	Description
ValueError	If packages fail validation
IOError	If file cannot be written

Source code in src/brix/modules/dbt/project/editor.py

def save_packages(packages: DbtPackages, project_dir: Path) -> None:
    """Save packages.yml to project directory.

    Args:
        packages: DbtPackages instance to save
        project_dir: Path to project directory (or dbt_project.yml file)

    Raises:
        ValueError: If packages fail validation
        IOError: If file cannot be written
    """
    logger = get_logger()

    # Handle both directory and file paths
    if project_dir.name == "dbt_project.yml":
        project_dir = project_dir.parent

    packages_path = project_dir / "packages.yml"

    # Validate by re-parsing
    yaml_content = packages.to_yaml()
    DbtPackages.from_yaml(yaml_content)

    # Write to disk
    packages_path.write_text(yaml_content)
    logger.debug("Saved packages to %s", packages_path)

update_project_field(project, field, value)

Update a single project field.

Parameters:

Name	Type	Description	Default
project	DbtProject	DbtProject instance	required
field	str	Field name to update	required
value	str | None	New value for the field	required

Returns:

Type	Description
DbtProject	Updated DbtProject instance

Raises:

Type	Description
InvalidFieldError	If field is not editable
ValueError	If value fails validation (e.g., invalid project name)

Source code in src/brix/modules/dbt/project/editor.py

def update_project_field(
    project: DbtProject,
    field: str,
    value: str | None,
) -> DbtProject:
    """Update a single project field.

    Args:
        project: DbtProject instance
        field: Field name to update
        value: New value for the field

    Returns:
        Updated DbtProject instance

    Raises:
        InvalidFieldError: If field is not editable
        ValueError: If value fails validation (e.g., invalid project name)
    """
    # Normalize field name (convert dashes to underscores)
    field = field.replace("-", "_")

    if field not in EDITABLE_FIELDS:
        msg = f"Field '{field}' is not editable. Editable fields: {', '.join(sorted(EDITABLE_FIELDS))}"
        raise InvalidFieldError(msg)

    # Special validation for project name
    if field == "name" and value is not None:
        validate_project_name(value)

    setattr(project, field, value)
    return project

update_path_field(project, field, action, value)

Update a path list field (add/remove/set).

Parameters:

Name	Type	Description	Default
project	DbtProject	DbtProject instance	required
field	str	Field name (model_paths, seed_paths, etc.)	required
action	Literal['add', 'remove', 'set']	Operation to perform	required
value	str | list[str]	Path(s) to add/remove, or full list for "set"	required

Returns:

Type	Description
DbtProject	Updated DbtProject instance

Raises:

Type	Description
InvalidFieldError	If field is not a path field
ValueError	If action is invalid or path not found for remove

Source code in src/brix/modules/dbt/project/editor.py

def update_path_field(
    project: DbtProject,
    field: str,
    action: Literal["add", "remove", "set"],
    value: str | list[str],
) -> DbtProject:
    """Update a path list field (add/remove/set).

    Args:
        project: DbtProject instance
        field: Field name (model_paths, seed_paths, etc.)
        action: Operation to perform
        value: Path(s) to add/remove, or full list for "set"

    Returns:
        Updated DbtProject instance

    Raises:
        InvalidFieldError: If field is not a path field
        ValueError: If action is invalid or path not found for remove
    """
    # Normalize field name (convert dashes to underscores)
    field = field.replace("-", "_")

    if field not in PATH_FIELDS:
        msg = f"Field '{field}' is not a path field. Path fields: {', '.join(sorted(PATH_FIELDS))}"
        raise InvalidFieldError(msg)

    current_paths: list[str] = getattr(project, field, [])

    if action == "add":
        path_to_add = value if isinstance(value, str) else value[0]
        if path_to_add not in current_paths:
            current_paths.append(path_to_add)
    elif action == "remove":
        path_to_remove = value if isinstance(value, str) else value[0]
        if path_to_remove not in current_paths:
            msg = f"Path '{path_to_remove}' not found in {field}"
            raise ValueError(msg)
        current_paths.remove(path_to_remove)
    elif action == "set":
        current_paths = list(value) if isinstance(value, list) else [value]
    else:
        msg = f"Invalid action: {action}. Must be 'add', 'remove', or 'set'"
        raise ValueError(msg)

    setattr(project, field, current_paths)
    return project

get_package_identifiers(packages)

Get list of all package identifiers for display.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required

Returns:

Type	Description
list[str]	List of package identifiers

Source code in src/brix/modules/dbt/project/editor.py

def get_package_identifiers(packages: DbtPackages) -> list[str]:
    """Get list of all package identifiers for display.

    Args:
        packages: DbtPackages instance

    Returns:
        List of package identifiers
    """
    return [_get_package_identifier(pkg) for pkg in packages.packages]

find_package_index(packages, identifier)

Find package index by identifier.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
identifier	str	Package name (hub), git URL, or local path	required

Returns:

Type	Description
int | None	Index of package or None if not found

Source code in src/brix/modules/dbt/project/editor.py

def find_package_index(packages: DbtPackages, identifier: str) -> int | None:
    """Find package index by identifier.

    Args:
        packages: DbtPackages instance
        identifier: Package name (hub), git URL, or local path

    Returns:
        Index of package or None if not found
    """
    for i, pkg in enumerate(packages.packages):
        if _get_package_identifier(pkg) == identifier:
            return i
    return None

has_package(packages, identifier)

Check if package exists.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
identifier	str	Package identifier	required

Returns:

Type	Description
bool	True if package exists

Source code in src/brix/modules/dbt/project/editor.py

def has_package(packages: DbtPackages, identifier: str) -> bool:
    """Check if package exists.

    Args:
        packages: DbtPackages instance
        identifier: Package identifier

    Returns:
        True if package exists
    """
    return find_package_index(packages, identifier) is not None

add_hub_package(packages, package_name, version)

Add a hub package.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
package_name	str	Package name (e.g., "dbt-labs/dbt_utils")	required
version	str	Version specifier (e.g., ">=1.0.0")	required

Returns:

Type	Description
DbtPackages	Updated DbtPackages instance

Raises:

Type	Description
PackageAlreadyExistsError	If package already exists

Source code in src/brix/modules/dbt/project/editor.py

def add_hub_package(
    packages: DbtPackages,
    package_name: str,
    version: str,
) -> DbtPackages:
    """Add a hub package.

    Args:
        packages: DbtPackages instance
        package_name: Package name (e.g., "dbt-labs/dbt_utils")
        version: Version specifier (e.g., ">=1.0.0")

    Returns:
        Updated DbtPackages instance

    Raises:
        PackageAlreadyExistsError: If package already exists
    """
    if has_package(packages, package_name):
        msg = f"Package '{package_name}' already exists"
        raise PackageAlreadyExistsError(msg)

    packages.packages.append(HubPackage(package=package_name, version=version))
    return packages

add_git_package(packages, git_url, revision, subdirectory=None)

Add a git package.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
git_url	str	Git repository URL	required
revision	str	Branch, tag, or commit hash	required
subdirectory	str | None	Optional subdirectory within repo	None

Returns:

Type	Description
DbtPackages	Updated DbtPackages instance

Raises:

Type	Description
PackageAlreadyExistsError	If package already exists

Source code in src/brix/modules/dbt/project/editor.py

def add_git_package(
    packages: DbtPackages,
    git_url: str,
    revision: str,
    subdirectory: str | None = None,
) -> DbtPackages:
    """Add a git package.

    Args:
        packages: DbtPackages instance
        git_url: Git repository URL
        revision: Branch, tag, or commit hash
        subdirectory: Optional subdirectory within repo

    Returns:
        Updated DbtPackages instance

    Raises:
        PackageAlreadyExistsError: If package already exists
    """
    if has_package(packages, git_url):
        msg = f"Git package '{git_url}' already exists"
        raise PackageAlreadyExistsError(msg)

    packages.packages.append(GitPackage(git=git_url, revision=revision, subdirectory=subdirectory))
    return packages

add_local_package(packages, local_path)

Add a local package.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
local_path	str	Local filesystem path	required

Returns:

Type	Description
DbtPackages	Updated DbtPackages instance

Raises:

Type	Description
PackageAlreadyExistsError	If package already exists

Source code in src/brix/modules/dbt/project/editor.py

def add_local_package(
    packages: DbtPackages,
    local_path: str,
) -> DbtPackages:
    """Add a local package.

    Args:
        packages: DbtPackages instance
        local_path: Local filesystem path

    Returns:
        Updated DbtPackages instance

    Raises:
        PackageAlreadyExistsError: If package already exists
    """
    if has_package(packages, local_path):
        msg = f"Local package '{local_path}' already exists"
        raise PackageAlreadyExistsError(msg)

    packages.packages.append(LocalPackage(local=local_path))
    return packages

remove_package(packages, identifier)

Remove a package by its identifier.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
identifier	str	Package name, git URL, or local path	required

Returns:

Type	Description
DbtPackages	Updated DbtPackages instance

Raises:

Type	Description
PackageNotFoundError	If package not found

Source code in src/brix/modules/dbt/project/editor.py

def remove_package(
    packages: DbtPackages,
    identifier: str,
) -> DbtPackages:
    """Remove a package by its identifier.

    Args:
        packages: DbtPackages instance
        identifier: Package name, git URL, or local path

    Returns:
        Updated DbtPackages instance

    Raises:
        PackageNotFoundError: If package not found
    """
    index = find_package_index(packages, identifier)
    if index is None:
        msg = f"Package '{identifier}' not found"
        raise PackageNotFoundError(msg)

    packages.packages.pop(index)
    return packages

update_package_version(packages, package_name, new_version)

Update a hub package version.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required
package_name	str	Package name (must be a hub package)	required
new_version	str	New version specifier	required

Returns:

Type	Description
DbtPackages	Updated DbtPackages instance

Raises:

Type	Description
PackageNotFoundError	If package not found
ValueError	If package is not a hub package

Source code in src/brix/modules/dbt/project/editor.py

def update_package_version(
    packages: DbtPackages,
    package_name: str,
    new_version: str,
) -> DbtPackages:
    """Update a hub package version.

    Args:
        packages: DbtPackages instance
        package_name: Package name (must be a hub package)
        new_version: New version specifier

    Returns:
        Updated DbtPackages instance

    Raises:
        PackageNotFoundError: If package not found
        ValueError: If package is not a hub package
    """
    index = find_package_index(packages, package_name)
    if index is None:
        msg = f"Package '{package_name}' not found"
        raise PackageNotFoundError(msg)

    pkg = packages.packages[index]
    if not isinstance(pkg, HubPackage):
        msg = f"Package '{package_name}' is not a hub package, cannot update version"
        raise ValueError(msg)

    pkg.version = new_version
    return packages

get_package_display_info(packages)

Get package information for display.

Parameters:

Name	Type	Description	Default
packages	DbtPackages	DbtPackages instance	required

Returns:

Type	Description
list[tuple[str, str]]	List of (identifier, type_info) tuples for display

Source code in src/brix/modules/dbt/project/editor.py

def get_package_display_info(packages: DbtPackages) -> list[tuple[str, str]]:
    """Get package information for display.

    Args:
        packages: DbtPackages instance

    Returns:
        List of (identifier, type_info) tuples for display
    """
    result: list[tuple[str, str]] = []
    for pkg in packages.packages:
        if isinstance(pkg, HubPackage):
            result.append((pkg.package, f"hub: {pkg.version}"))
        elif isinstance(pkg, GitPackage):
            info = f"git: {pkg.revision}"
            if pkg.subdirectory:
                info += f" ({pkg.subdirectory})"
            result.append((pkg.git, info))
        else:
            result.append((pkg.local, "local"))
    return result

Project Finder

brix.modules.dbt.project.finder

Project discovery service for dbt projects.

Provides functions to find dbt_project.yml files in a directory tree with interactive fuzzy selection support.

get_search_root()

Get the search root directory.

Returns git repository root if in a git repo, otherwise current working directory.

Returns:

Type	Description
Path	Path to search root directory

Source code in src/brix/modules/dbt/project/finder.py

def get_search_root() -> Path:
    """Get the search root directory.

    Returns git repository root if in a git repo, otherwise current working directory.

    Returns:
        Path to search root directory
    """
    logger = get_logger()

    try:
        result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],  # noqa: S607
            capture_output=True,
            text=True,
            check=True,
        )
        git_root = Path(result.stdout.strip())
        logger.debug("Using git root as search root: %s", git_root)
        return git_root
    except (subprocess.CalledProcessError, FileNotFoundError):
        cwd = Path.cwd()
        logger.debug("Not in git repo, using cwd as search root: %s", cwd)
        return cwd

find_dbt_projects(root=None, max_depth=10)

Find all dbt_project.yml files under root directory.

Parameters:

Name	Type	Description	Default
root	Path | None	Search root (uses get_search_root() if None)	None
max_depth	int	Maximum directory depth to search (default 10)	10

Returns:

Type	Description
list[Path]	List of absolute paths to dbt_project.yml files, sorted by path

Source code in src/brix/modules/dbt/project/finder.py

def find_dbt_projects(
    root: Path | None = None,
    max_depth: int = 10,
) -> list[Path]:
    """Find all dbt_project.yml files under root directory.

    Args:
        root: Search root (uses get_search_root() if None)
        max_depth: Maximum directory depth to search (default 10)

    Returns:
        List of absolute paths to dbt_project.yml files, sorted by path
    """
    logger = get_logger()
    search_root = root or get_search_root()

    if not search_root.exists():
        logger.warning("Search root does not exist: %s", search_root)
        return []

    projects: list[Path] = []

    # Use glob to find all dbt_project.yml files
    for project_file in search_root.glob("**/dbt_project.yml"):
        # Check depth
        try:
            relative = project_file.relative_to(search_root)
            depth = len(relative.parts) - 1  # Subtract 1 for the filename itself
            if depth > max_depth:
                continue
        except ValueError:
            continue

        # Check exclusions
        if _should_exclude(project_file):
            logger.debug("Excluding project in excluded directory: %s", project_file)
            continue

        projects.append(project_file.resolve())
        logger.debug("Found dbt project: %s", project_file)

    # Sort by path for consistent ordering
    projects.sort()
    logger.debug("Found %d dbt projects", len(projects))

    return projects

prompt_select_project(projects, search_root=None)

Interactive project selection with fuzzy autocomplete.

Parameters:

Name	Type	Description	Default
projects	list[Path]	List of dbt_project.yml paths	required
search_root	Path | None	Root directory for relative path display (uses get_search_root() if None)	None

Returns:

Type	Description
Path | None	Selected project path, or None if cancelled

Source code in src/brix/modules/dbt/project/finder.py

def prompt_select_project(
    projects: list[Path],
    search_root: Path | None = None,
) -> Path | None:
    """Interactive project selection with fuzzy autocomplete.

    Args:
        projects: List of dbt_project.yml paths
        search_root: Root directory for relative path display (uses get_search_root() if None)

    Returns:
        Selected project path, or None if cancelled
    """
    if not projects:
        return None

    root = search_root or get_search_root()

    # Build choice mapping: display string -> actual path
    choices: dict[str, Path] = {}
    for project_path in projects:
        display = _format_project_choice(project_path, root)
        # Handle duplicate display names by appending parent info
        if display in choices:
            display = str(project_path.parent)
        choices[display] = project_path

    # Use autocomplete for fuzzy search if many projects, otherwise select
    if len(choices) > 5:
        selected = questionary.autocomplete(
            "Select project (type to filter):",
            choices=list(choices.keys()),
            match_middle=True,
        ).ask()
    else:
        selected = questionary.select(
            "Select project:",
            choices=list(choices.keys()),
        ).ask()

    if selected is None:
        return None

    return choices.get(selected)

discover_and_select_project(root=None, max_depth=10)

Combined discovery and selection flow.

Finds dbt projects in the directory tree, prompts user to select one, and loads the selected project.

Parameters:

Name	Type	Description	Default
root	Path | None	Search root (uses get_search_root() if None)	None
max_depth	int	Maximum directory depth to search	10

Returns:

Type	Description
tuple[Path, DbtProject] | None	Tuple of (project_path, loaded DbtProject) or None if cancelled/not found

Source code in src/brix/modules/dbt/project/finder.py

def discover_and_select_project(
    root: Path | None = None,
    max_depth: int = 10,
) -> tuple[Path, DbtProject] | None:
    """Combined discovery and selection flow.

    Finds dbt projects in the directory tree, prompts user to select one,
    and loads the selected project.

    Args:
        root: Search root (uses get_search_root() if None)
        max_depth: Maximum directory depth to search

    Returns:
        Tuple of (project_path, loaded DbtProject) or None if cancelled/not found
    """
    import typer

    search_root = root or get_search_root()
    projects = find_dbt_projects(search_root, max_depth)

    if not projects:
        typer.echo(f"No dbt projects found under {search_root}", err=True)
        return None

    if len(projects) == 1:
        # Only one project found, use it directly
        project_path = projects[0]
        typer.echo(f"Found project: {project_path.parent}")
    else:
        typer.echo(f"Found {len(projects)} dbt projects")
        project_path = prompt_select_project(projects, search_root)
        if project_path is None:
            return None

    # Load the project
    try:
        project = DbtProject.from_file(project_path)
        return (project_path, project)
    except Exception as e:
        typer.echo(f"Error loading project: {e}", err=True)
        return None

dbt Passthrough

brix.modules.dbt.passthrough

dbt module - business logic for dbt operations.

DbtNotFoundError

Bases: Exception

Raised when dbt executable cannot be found.

Source code in src/brix/modules/dbt/passthrough.py

class DbtNotFoundError(Exception):
    """Raised when dbt executable cannot be found."""

ProjectPathCache

Bases: BaseModel

Cached project path for dbt passthrough.

Source code in src/brix/modules/dbt/passthrough.py

class ProjectPathCache(BaseModel):
    """Cached project path for dbt passthrough."""

    project_path: Path

CachedPathNotFoundError

Bases: FileNotFoundError

Raised when cached project path no longer exists.

Source code in src/brix/modules/dbt/passthrough.py

class CachedPathNotFoundError(FileNotFoundError):
    """Raised when cached project path no longer exists."""

load_project_cache()

Load cached project path.

Returns:

Type	Description
Path | None	Cached project path if valid, None otherwise.

Raises:

Type	Description
CachedPathNotFoundError	If cached path no longer exists or is not a directory.

Source code in src/brix/modules/dbt/passthrough.py

def load_project_cache() -> Path | None:
    """Load cached project path.

    Returns:
        Cached project path if valid, None otherwise.

    Raises:
        CachedPathNotFoundError: If cached path no longer exists or is not a directory.
    """
    logger = get_logger()
    if not PROJECT_CACHE_FILE.exists():
        logger.debug("Project cache file not found: %s", PROJECT_CACHE_FILE)
        return None
    try:
        cache = ProjectPathCache.model_validate_json(PROJECT_CACHE_FILE.read_text())
    except (ValidationError, OSError) as e:
        logger.debug("Failed to load project cache: %s", e)
        return None

    # Check if cached path still exists (outside try/except to propagate error)
    if not cache.project_path.exists():
        logger.debug("Cached project path no longer exists: %s", cache.project_path)
        raise CachedPathNotFoundError(f"Cached project path no longer exists: {cache.project_path}")
    if not cache.project_path.is_dir():
        logger.debug("Cached project path is not a directory: %s", cache.project_path)
        raise CachedPathNotFoundError(f"Cached project path is not a directory: {cache.project_path}")

    logger.debug("Loaded project cache: %s", cache.project_path)
    return cache.project_path

save_project_cache(project_path)

Save project path to cache.

Converts relative paths to absolute before saving.

Parameters:

Name	Type	Description	Default
project_path	Path	The project path to cache.	required

Source code in src/brix/modules/dbt/passthrough.py

def save_project_cache(project_path: Path) -> None:
    """Save project path to cache.

    Converts relative paths to absolute before saving.

    Args:
        project_path: The project path to cache.
    """
    logger = get_logger()
    absolute_path = project_path.resolve()
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    cache = ProjectPathCache(project_path=absolute_path)
    PROJECT_CACHE_FILE.write_text(cache.model_dump_json())
    logger.debug("Project cache saved: %s", absolute_path)

find_dbt_executable()

Find the dbt executable path.

This function handles two scenarios: 1. brix is installed in the same venv as dbt - dbt should be directly available 2. brix is installed as a global tool - need to discover project venv with dbt

Returns:

Type	Description
str	Path to the dbt executable.

Raises:

Type	Description
DbtNotFoundError	If dbt cannot be found.

Source code in src/brix/modules/dbt/passthrough.py

def find_dbt_executable() -> str:
    """Find the dbt executable path.

    This function handles two scenarios:
    1. brix is installed in the same venv as dbt - dbt should be directly available
    2. brix is installed as a global tool - need to discover project venv with dbt

    Returns:
        Path to the dbt executable.

    Raises:
        DbtNotFoundError: If dbt cannot be found.
    """
    # TODO: Implement venv discovery logic for when brix is installed globally
    # but dbt is in a project-specific venv. This could involve:
    # - Looking for .venv/ in current directory or parent directories
    # - Checking for pyproject.toml/requirements.txt to identify project root
    # - Activating the discovered venv or returning path to its dbt executable

    # For now, assume dbt is available in PATH (same venv scenario)
    return "dbt"

pre_dbt_hook()

Hook for setup before running dbt. Placeholder for future logic.

Source code in src/brix/modules/dbt/passthrough.py

def pre_dbt_hook() -> None:
    """Hook for setup before running dbt. Placeholder for future logic."""
    # TODO: Placeholder for other stuff that has to happen before running dbt.
    pass

run_dbt(args, project_path=None)

Run dbt with the given arguments and return exit code.

Parameters:

Name	Type	Description	Default
args	list[str]	List of arguments to pass to dbt.	required
project_path	Path | None	Optional directory to run dbt in.	None

Returns:

Type	Description
int	Exit code from the dbt process (1 if dbt not found or invalid project path).

Source code in src/brix/modules/dbt/passthrough.py

def run_dbt(args: list[str], project_path: Path | None = None) -> int:
    """Run dbt with the given arguments and return exit code.

    Args:
        args: List of arguments to pass to dbt.
        project_path: Optional directory to run dbt in.

    Returns:
        Exit code from the dbt process (1 if dbt not found or invalid project path).
    """
    logger = get_logger()

    pre_dbt_hook()

    # Validate project path if provided
    if project_path is not None:
        if not project_path.exists():
            logger.error("Project path does not exist: %s", project_path)
            return 1
        if not project_path.is_dir():
            logger.error("Project path is not a directory: %s", project_path)
            return 1

    try:
        dbt_path = find_dbt_executable()
    except DbtNotFoundError as e:
        logger.error(str(e))
        return 1

    cwd = project_path.resolve() if project_path else None
    logger.debug("Executing dbt command: %s %s (cwd=%s)", dbt_path, " ".join(args), cwd)
    try:
        # "unsafe" passthrough is intended, we trust the user to pass valid arguments. Its their local machine.
        result = subprocess.run([dbt_path, *args], cwd=cwd)  # noqa: S603
    except FileNotFoundError:
        logger.error(
            "dbt not found in PATH. Ensure dbt is installed and available.\n"
            "If using a virtual environment, make sure it's activated or install brix in the same environment as dbt."
        )
        return 1
    except OSError as e:
        logger.error("Failed to execute dbt: %s", e)
        return 1

    logger.debug("dbt exited with code: %d", result.returncode)

    return result.returncode

Logging

brix.utils.logging

Terraform-style logging system for brix CLI.

Environment variables

BRIX_LOG: Log level (TRACE, DEBUG, INFO, WARN, ERROR, OFF) BRIX_LOG_PATH: File path for log output BRIX_LOG_JSON: Enable JSON format (true/false)

Logging convention: Use %-formatting for logger calls (lazy evaluation), f-strings elsewhere. This is Python logging best practice.

LogLevel

Bases: IntEnum

Log levels matching Terraform's TF_LOG.

Custom TRACE level added below DEBUG (Python's DEBUG=10).

Source code in src/brix/utils/logging.py

class LogLevel(IntEnum):
    """Log levels matching Terraform's TF_LOG.

    Custom TRACE level added below DEBUG (Python's DEBUG=10).
    """

    TRACE = 5
    DEBUG = 10
    INFO = 20
    WARN = 30
    ERROR = 40
    OFF = 100

LogConfig

Bases: BaseSettings

Logging configuration from environment variables.

Environment variables

BRIX_LOG: Log level (TRACE, DEBUG, INFO, WARN, ERROR, OFF) BRIX_LOG_PATH: File path for log output BRIX_LOG_JSON: Enable JSON format (true/false)

Source code in src/brix/utils/logging.py

class LogConfig(BaseSettings):
    """Logging configuration from environment variables.

    Environment variables:
        BRIX_LOG: Log level (TRACE, DEBUG, INFO, WARN, ERROR, OFF)
        BRIX_LOG_PATH: File path for log output
        BRIX_LOG_JSON: Enable JSON format (true/false)
    """

    model_config = SettingsConfigDict(
        env_prefix="BRIX_",
        case_sensitive=False,
    )

    log: Literal["TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF"] = "ERROR"
    log_path: Path | None = None
    log_json: bool = False

    @field_validator("log", mode="before")
    @classmethod
    def normalize_log_level(cls, v: str) -> str:
        """Normalize log level to uppercase."""
        if isinstance(v, str):
            v = v.upper()
            # Handle WARNING -> WARN alias
            if v == "WARNING":
                return "WARN"
        return v

normalize_log_level(v) classmethod

Normalize log level to uppercase.

Source code in src/brix/utils/logging.py

@field_validator("log", mode="before")
@classmethod
def normalize_log_level(cls, v: str) -> str:
    """Normalize log level to uppercase."""
    if isinstance(v, str):
        v = v.upper()
        # Handle WARNING -> WARN alias
        if v == "WARNING":
            return "WARN"
    return v

BrixFormatter

Bases: Formatter

Human-readable log formatter for console output.

Format: [2024-01-15T10:30:45Z] [DEBUG] message

Source code in src/brix/utils/logging.py

class BrixFormatter(logging.Formatter):
    """Human-readable log formatter for console output.

    Format: ``[2024-01-15T10:30:45Z] [DEBUG] message``
    """

    def format(self, record: logging.LogRecord) -> str:
        """Format the log record as human-readable text."""
        timestamp = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat()
        level = record.levelname
        return f"[{timestamp}] [{level}] {record.getMessage()}"

format(record)

Format the log record as human-readable text.

Source code in src/brix/utils/logging.py

def format(self, record: logging.LogRecord) -> str:
    """Format the log record as human-readable text."""
    timestamp = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat()
    level = record.levelname
    return f"[{timestamp}] [{level}] {record.getMessage()}"

BrixJsonFormatter

Bases: Formatter

JSON log formatter for file output and machine parsing.

Output: {"@timestamp": "...", "@level": "DEBUG", "@message": "...", "@module": "..."}

Source code in src/brix/utils/logging.py

class BrixJsonFormatter(logging.Formatter):
    """JSON log formatter for file output and machine parsing.

    Output: {"@timestamp": "...", "@level": "DEBUG", "@message": "...", "@module": "..."}
    """

    def format(self, record: logging.LogRecord) -> str:
        """Format the log record as JSON."""
        log_obj: dict[str, Any] = {
            "@timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "@level": record.levelname,
            "@message": record.getMessage(),
            "@module": record.module,
        }
        if record.exc_info:
            log_obj["@exception"] = self.formatException(record.exc_info)
        return json.dumps(log_obj)

format(record)

Format the log record as JSON.

Source code in src/brix/utils/logging.py

def format(self, record: logging.LogRecord) -> str:
    """Format the log record as JSON."""
    log_obj: dict[str, Any] = {
        "@timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
        "@level": record.levelname,
        "@message": record.getMessage(),
        "@module": record.module,
    }
    if record.exc_info:
        log_obj["@exception"] = self.formatException(record.exc_info)
    return json.dumps(log_obj)

setup_logging(level=None, log_path=None, json_format=None)

Initialize the brix logger with config from env and CLI overrides.

CLI arguments override environment variables. Thread-safe initialization with singleton pattern.

Parameters:

Name	Type	Description	Default
level	str | None	CLI override for BRIX_LOG	None
log_path	Path | None	CLI override for BRIX_LOG_PATH	None
json_format	bool | None	CLI override for BRIX_LOG_JSON	None

Returns:

Type	Description
Logger	Configured logger instance

Source code in src/brix/utils/logging.py

def setup_logging(
    level: str | None = None,
    log_path: Path | None = None,
    json_format: bool | None = None,
) -> logging.Logger:
    """Initialize the brix logger with config from env and CLI overrides.

    CLI arguments override environment variables.
    Thread-safe initialization with singleton pattern.

    Args:
        level: CLI override for BRIX_LOG
        log_path: CLI override for BRIX_LOG_PATH
        json_format: CLI override for BRIX_LOG_JSON

    Returns:
        Configured logger instance
    """
    global _logger

    with _lock:
        if _logger is not None:
            return _logger

        # Load config from env vars
        config = LogConfig()

        # Apply CLI overrides
        effective_level = level.upper() if level else config.log
        effective_path = log_path if log_path is not None else config.log_path
        effective_json = json_format if json_format is not None else config.log_json

        # Create logger
        _logger = logging.getLogger("brix")
        _logger.setLevel(LogLevel[effective_level].value)
        _logger.handlers.clear()  # Prevent duplicate handlers

        if effective_level == "OFF":
            _logger.addHandler(logging.NullHandler())
            return _logger

        # Console handler (stderr, human-readable by default)
        console_handler = logging.StreamHandler()
        if effective_json and not effective_path:
            # JSON to console only if no file path and JSON requested
            console_handler.setFormatter(BrixJsonFormatter())
        else:
            console_handler.setFormatter(BrixFormatter())
        _logger.addHandler(console_handler)

        # File handler (JSON by default for machine parsing, unless explicitly disabled)
        if effective_path:
            file_handler = logging.FileHandler(effective_path)
            # Use JSON for files unless json_format was explicitly set to False
            use_json_for_file = json_format is not False
            if use_json_for_file:
                file_handler.setFormatter(BrixJsonFormatter())
            else:
                file_handler.setFormatter(BrixFormatter())
            _logger.addHandler(file_handler)

        return _logger

get_logger()

Get the brix logger (initializes with defaults if needed).

For use throughout the codebase.

Source code in src/brix/utils/logging.py

def get_logger() -> logging.Logger:
    """Get the brix logger (initializes with defaults if needed).

    For use throughout the codebase.
    """
    global _logger
    if _logger is None:
        return setup_logging()
    return _logger

reset_logger()

Reset the logger singleton (for testing).

Source code in src/brix/utils/logging.py

def reset_logger() -> None:
    """Reset the logger singleton (for testing)."""
    global _logger
    with _lock:
        if _logger is not None:
            _logger.handlers.clear()
            _logger = None