Errors

PyCharter's exception hierarchy for error handling.

Exception Hierarchy

PyCharterError (base)
├── ConfigError
│   └── ConfigLoadError
├── ConfigValidationError
└── ExpressionError

PyCharterError

Base exception for all PyCharter errors:

from pycharter.shared.errors import PyCharterError

try:
    result = await pipeline.run()
except PyCharterError as e:
    print(f"PyCharter error: {e}")

ConfigError

Raised for configuration loading/parsing failures:

from pycharter.shared.errors import ConfigError

try:
    pipeline = Pipeline.from_config_dir("invalid/path/")
except ConfigError as e:
    print(f"Config error: {e}")

ConfigValidationError

Raised when configuration fails schema validation:

from pycharter.shared.errors import ConfigValidationError

try:
    pipeline = Pipeline.from_config_files(
        extract="extract.yaml",  # Missing required 'type' field
        load="load.yaml"
    )
except ConfigValidationError as e:
    print(f"Validation error: {e}")

ConfigLoadError

Raised when a config file cannot be loaded:

from pycharter.shared.errors import ConfigLoadError

try:
    pipeline = Pipeline.from_config_files(
        extract="nonexistent.yaml",
        load="load.yaml"
    )
except ConfigLoadError as e:
    print(f"Load error: {e}")

ExpressionError

Raised when expression evaluation fails:

from pycharter.shared.errors import ExpressionError

try:
    transform = AddField("result", "invalid_function()")
except ExpressionError as e:
    print(f"Expression error: {e}")

ErrorMode

Control how pipeline errors are handled:

from pycharter.shared.errors import ErrorMode, ErrorContext

# STRICT: Raise on first error (default)
result = await pipeline.run(
    error_context=ErrorContext(mode=ErrorMode.STRICT)
)

# LENIENT: Log and continue
result = await pipeline.run(
    error_context=ErrorContext(mode=ErrorMode.LENIENT)
)

# COLLECT: Gather all errors
result = await pipeline.run(
    error_context=ErrorContext(mode=ErrorMode.COLLECT)
)
print(result.errors)

ErrorContext

ErrorContext dataclass

ErrorContext(
    mode: ErrorMode = LENIENT,
    errors: list[str] = list(),
    warnings: list[str] = list(),
)

Context for tracking errors during operations.

Use with ErrorMode.COLLECT to gather errors without stopping.

has_errors property

has_errors: bool

Whether any errors were collected.

handle_error

handle_error(
    message: str,
    exception: Exception | None = None,
    category: str = "error",
) -> None

Handle an error according to the current mode.

Parameters:

Name	Type	Description	Default
message	str	Error message	required
exception	Exception | None	Optional exception that caused the error	None
category	str	Error category for logging	'error'

Source code in src/pycharter/shared/errors.py

def handle_error(
    self,
    message: str,
    exception: Exception | None = None,
    category: str = "error",
) -> None:
    """
    Handle an error according to the current mode.

    Args:
        message: Error message
        exception: Optional exception that caused the error
        category: Error category for logging
    """
    full_message = f"{category}: {message}"
    if exception:
        full_message += f" ({type(exception).__name__}: {exception})"

    if self.mode == ErrorMode.STRICT:
        if exception:
            raise type(exception)(full_message) from exception
        raise ValueError(full_message)

    elif self.mode == ErrorMode.LENIENT:
        warnings.warn(full_message)
        logger.warning(full_message)
        self.warnings.append(full_message)

    elif self.mode == ErrorMode.COLLECT:
        self.errors.append(full_message)
        logger.warning(full_message)

handle_warning

handle_warning(
    message: str, category: str = "warning"
) -> None

Handle a warning (non-fatal issue).

Parameters:

Name	Type	Description	Default
message	str	Warning message	required
category	str	Warning category	'warning'

Source code in src/pycharter/shared/errors.py

def handle_warning(self, message: str, category: str = "warning") -> None:
    """
    Handle a warning (non-fatal issue).

    Args:
        message: Warning message
        category: Warning category
    """
    full_message = f"{category}: {message}"
    warnings.warn(full_message)
    logger.warning(full_message)
    self.warnings.append(full_message)

raise_if_errors

raise_if_errors() -> None

Raise ValueError if any errors were collected.

Source code in src/pycharter/shared/errors.py

def raise_if_errors(self) -> None:
    """Raise ValueError if any errors were collected."""
    if self.errors:
        raise ValueError(f"Errors occurred: {'; '.join(self.errors)}")

clear

clear() -> None

Clear all errors and warnings.

Source code in src/pycharter/shared/errors.py

def clear(self) -> None:
    """Clear all errors and warnings."""
    self.errors.clear()
    self.warnings.clear()

Best Practices

Catch Specific Exceptions

from pycharter.shared.errors import (
    PyCharterError,
    ConfigError,
    ConfigValidationError,
    ExpressionError
)

try:
    pipeline = Pipeline.from_config_dir("pipelines/")
    result = await pipeline.run()
except ConfigValidationError as e:
    # Handle invalid config
    print(f"Invalid configuration: {e}")
except ConfigError as e:
    # Handle missing/corrupt config
    print(f"Configuration error: {e}")
except ExpressionError as e:
    # Handle expression errors
    print(f"Expression error: {e}")
except PyCharterError as e:
    # Catch-all for other PyCharter errors
    print(f"PyCharter error: {e}")

Use ErrorContext for Batch Processing

# Collect errors for batch analysis
error_context = ErrorContext(mode=ErrorMode.COLLECT)
result = await pipeline.run(error_context=error_context)

if result.errors:
    # Log all errors
    for error in result.errors:
        logger.error(f"Pipeline error: {error}")

    # Decide whether to fail
    if len(result.errors) > 10:
        raise Exception(f"Too many errors: {len(result.errors)}")