tessl/pypi-dbt-core

A command-line tool for data transformation and analytics engineering workflows.

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-dbt-core
Author: tessl

dbt-core provides a comprehensive exception hierarchy for structured error handling and reporting. Most exceptions inherit from the dbt-common exception system and include error codes, structured messaging, and context information.

Core Exception Classes

dbt-core Specific Exceptions

class ContractBreakingChangeError(DbtRuntimeError):
    """
    Breaking change to an enforced contract.
    
    Raised when dbt detects changes that break enforced model contracts
    during state comparison.
    """
    
    CODE = 10016
    MESSAGE = "Breaking Change to Contract"
    
    def __init__(
        self,
        breaking_changes: List[str],
        node=None,
    ) -> None:
        """
        Initialize contract breaking change error.
        
        Args:
            breaking_changes: List of specific breaking changes detected
            node: Node where the breaking change occurred (optional)
        """
    
    @property
    def type(self):
        """Error type description."""
        return "Breaking change to contract"
    
    def message(self):
        """Formatted error message with breaking change details."""

dbt-common Base Exceptions

These exceptions are imported from dbt-common and form the foundation of dbt's error system:

class DbtInternalError(Exception):
    """
    Internal dbt errors indicating bugs or unexpected conditions.
    
    These represent errors in dbt's logic rather than user configuration
    or data issues.
    """

class DbtRuntimeError(Exception):
    """
    Runtime execution errors during dbt operations.
    
    These occur during execution and typically indicate issues with
    SQL execution, database connectivity, or data processing.
    """

class DbtConfigError(Exception):
    """
    Configuration-related errors.
    
    Raised when there are issues with project configuration, profiles,
    or other configuration files.
    """

class DbtValidationError(Exception):
    """
    Data validation errors.
    
    Raised when data doesn't meet expected validation criteria or
    schema requirements.
    """

class CompilationError(Exception):
    """
    SQL compilation errors.
    
    Raised when dbt cannot compile models, macros, or other SQL
    into valid executable statements.
    """

class CommandResultError(Exception):
    """
    Command execution errors.
    
    Raised when external commands or operations fail during
    dbt execution.
    """

CLI-Specific Exceptions

dbt.cli.exceptions

class DbtUsageException(Exception):
    """
    CLI usage errors.
    
    Raised when command-line arguments are invalid or when
    CLI commands are used incorrectly.
    """

class DbtInternalException(Exception):
    """
    Internal CLI errors.
    
    Raised for unexpected internal errors in the CLI layer.
    """

Exception Properties

Most dbt exceptions include structured information:

Error Codes

# Each exception type has a unique error code
exception.CODE  # Integer error code for programmatic handling

Error Messages

# Structured error messages
exception.MESSAGE  # Base message template
exception.message()  # Formatted message with context
str(exception)  # String representation

Context Information

# Many exceptions include context about where the error occurred
exception.node  # dbt node where error occurred (if applicable)  
exception.type  # Error type description

Usage Patterns

Basic Exception Handling

from dbt.cli.main import dbtRunner
from dbt.exceptions import (
    DbtRuntimeError,
    CompilationError,
    DbtConfigError
)

runner = dbtRunner()

try:
    result = runner.invoke(['run'])
    if not result.success:
        raise result.exception
except CompilationError as e:
    print(f"Compilation failed: {e}")
    # Handle SQL compilation issues
except DbtConfigError as e:
    print(f"Configuration error: {e}")
    # Handle configuration problems
except DbtRuntimeError as e:
    print(f"Runtime error: {e}")
    # Handle execution issues

CLI Exception Handling

from dbt.cli.exceptions import DbtUsageException, DbtInternalException
from click.exceptions import ClickException

try:
    result = runner.invoke(['run', '--invalid-flag'])
except DbtUsageException as e:
    print(f"Usage error: {e}")
    # Handle invalid CLI usage
except DbtInternalException as e:
    print(f"Internal CLI error: {e}")
    # Handle internal CLI issues
except ClickException as e:
    print(f"Click error: {e}")
    # Handle Click framework errors

Contract Validation

from dbt.exceptions import ContractBreakingChangeError

try:
    result = runner.invoke(['run', '--select', 'state:modified'])
except ContractBreakingChangeError as e:
    print(f"Contract validation failed:")
    print(f"Error type: {e.type}")
    for change in e.breaking_changes:
        print(f"  - {change}")
    
    # Access error details
    print(f"Error code: {e.CODE}")
    print(f"Node: {e.node}")

Error Code Handling

def handle_dbt_error(exception):
    """Handle dbt errors based on error codes."""
    if hasattr(exception, 'CODE'):
        code = exception.CODE
        
        if code == 10016:  # ContractBreakingChangeError
            print("Handle contract breaking change")
        elif code in [1001, 1002]:  # Compilation errors
            print("Handle compilation issues")
        else:
            print(f"Handle error code {code}")
    else:
        print("Handle generic error")

try:
    result = runner.invoke(['compile'])
except Exception as e:
    handle_dbt_error(e)

Logging Errors

import logging
from dbt.exceptions import DbtRuntimeError

logger = logging.getLogger(__name__)

try:
    result = runner.invoke(['test'])
except DbtRuntimeError as e:
    # Log structured error information
    logger.error(
        "dbt test failed",
        extra={
            'error_code': getattr(e, 'CODE', None),
            'error_type': getattr(e, 'type', type(e).__name__),
            'node': getattr(e, 'node', None),
            'message': str(e)
        }
    )
    raise

Error Categories

Configuration Errors

Invalid dbt_project.yml syntax
Missing or invalid profiles.yml
Invalid package specifications
Schema validation failures

from dbt.exceptions import DbtConfigError

try:
    runner.invoke(['parse'])
except DbtConfigError as e:
    print("Fix configuration issues")

Compilation Errors

Invalid Jinja syntax in models
Undefined macros or variables
Circular dependencies
Invalid SQL syntax

from dbt.exceptions import CompilationError

try:
    runner.invoke(['compile'])
except CompilationError as e:
    print("Fix SQL or Jinja issues")

Runtime Errors

Database connection failures
SQL execution errors
Permission issues
Resource conflicts

from dbt.exceptions import DbtRuntimeError

try:
    runner.invoke(['run'])
except DbtRuntimeError as e:
    print("Fix runtime execution issues")

Validation Errors

Schema test failures
Data quality check failures
Contract validation failures

from dbt.exceptions import DbtValidationError, ContractBreakingChangeError

try:
    runner.invoke(['test'])
except ContractBreakingChangeError as e:
    print("Contract validation failed")
except DbtValidationError as e:
    print("Data validation failed")

Integration with External Systems

Airflow Integration

from airflow.exceptions import AirflowException
from dbt.exceptions import DbtRuntimeError

def airflow_dbt_task():
    try:
        result = runner.invoke(['run'])
        if not result.success:
            raise AirflowException(f"dbt failed: {result.exception}")
    except DbtRuntimeError as e:
        # Convert to Airflow exception
        raise AirflowException(f"dbt runtime error: {e}")

Custom Error Handling

class CustomDbtError(Exception):
    """Custom exception for application-specific dbt errors."""
    
    def __init__(self, original_error, context=None):
        self.original_error = original_error
        self.context = context or {}
        super().__init__(str(original_error))

def run_with_custom_handling():
    try:
        result = runner.invoke(['run'])
    except Exception as e:
        raise CustomDbtError(
            e, 
            context={'operation': 'run', 'timestamp': time.time()}
        )

Install with Tessl CLI