base

base

Base models and shared functionality.

Classes

Base

Base(**data)

Bases: BaseModel

Custom base model that

• Collects non-breaking validation or runtime warnings using WarningCollector
• Fails fast on critical errors
• Catches validation errors and converts them into warnings
• Validates on assignment, converting assignment errors into warnings
• Provides serialization to DataFrame

Initialize the model, converting validation errors to warnings.

Source code in src/pyetm/models/base.py

def __init__(self, **data: Any) -> None:
    """
    Initialize the model, converting validation errors to warnings.
    """
    super(BaseModel, self).__setattr__("__pydantic_private__", {})

    # Initialize all private attributes with their defaults
    private_dict: Dict[str, Any] = self.__pydantic_private__  # type: ignore[assignment]
    for attr_name, attr_info in self.__class__.__private_attributes__.items():
        if (
            hasattr(attr_info, "default_factory")
            and attr_info.default_factory is not None
        ):
            # Call factory - signature varies between pydantic versions
            private_dict[attr_name] = attr_info.default_factory()
        elif hasattr(attr_info, "default"):
            private_dict[attr_name] = attr_info.default
        else:
            private_dict[attr_name] = None

    try:
        super().__init__(**data)
    except ValidationError as e:
        # Check if data is None or empty - this indicates API error
        if not data or data is None:
            # Re-raise with clearer message
            raise ValueError(
                f"Cannot create {self.__class__.__name__} with empty data. "
                "This usually indicates an authentication or API error."
            ) from e

        # If validation fails, create model without validation and collect warnings
        # Use model_construct to bypass validation
        temp_instance = self.__class__.model_construct(**data)

        # Copy the constructed data to this instance
        for field_name, field_value in temp_instance.__dict__.items():
            if not field_name.startswith("_"):
                object.__setattr__(self, field_name, field_value)

        # Ensure required Pydantic slot attributes exist to prevent AttributeError
        for slot in ("__pydantic_fields_set__", "__pydantic_extra__"):
            try:
                value = object.__getattribute__(temp_instance, slot)
                object.__setattr__(self, slot, value)
            except AttributeError:
                # Initialize missing slot attributes with defaults
                if slot == "__pydantic_extra__":
                    object.__setattr__(self, slot, {})
                elif slot == "__pydantic_fields_set__":
                    object.__setattr__(self, slot, set())

        # Convert validation errors to warnings
        for error in e.errors():
            field_path = ".".join(str(part) for part in error.get("loc", []))
            message = error.get("msg", "Validation failed")
            self._warning_collector.add(field_path, message, "error")

Attributes

warnings property

warnings

Return warnings.

Functions

__setattr__

__setattr__(name, value)

Handle assignment with validation error capture.

Source code in src/pyetm/models/base.py

def __setattr__(self, name: str, value: Any) -> None:
    """
    Handle assignment with validation error capture.
    """
    # Skip validation for private attributes, methods/functions, or existing methods
    if (
        name.startswith("_")
        or name not in self.__class__.model_fields
        or callable(value)
        or hasattr(self.__class__, name)
    ):
        # Use object.__setattr__ to bypass Pydantic for these cases
        object.__setattr__(self, name, value)
        return

    # Clear existing warnings for this field
    self._warning_collector.clear(name)

    try:
        # Try to validate the new value by creating a copy with the update
        current_data = self.model_dump()
        current_data[name] = value

        # Test validation with a temporary instance
        test_instance = self.__class__.model_validate(current_data)

        # If validation succeeds, set the value
        super().__setattr__(name, value)

    except ValidationError as e:
        # If validation fails, add warnings but don't set the value
        for error in e.errors():
            if error.get("loc") == (name,):
                message = error.get("msg", "Validation failed")
                self._warning_collector.add(name, message, "warning")
        return

add_warning

add_warning(field, message, severity='warning')

Add a warning to this model instance.

Parameters:

Name	Type	Description	Default
field	str	Field name where warning occurred	required
message	Union[str, List[str], Dict[str, Any]]	Warning message (str, list of strings, or dict)	required
severity	Literal['info', 'warning', 'error']	Warning severity level (info/warning/error)	'warning'

Raises:

Type	Description
RuntimeError	If error policy determines warning should raise an exception

Source code in src/pyetm/models/base.py

def add_warning(
    self,
    field: str,
    message: Union[str, List[str], Dict[str, Any]],
    severity: Literal["info", "warning", "error"] = "warning",
) -> None:
    """Add a warning to this model instance.

    Args:
        field: Field name where warning occurred
        message: Warning message (str, list of strings, or dict)
        severity: Warning severity level (info/warning/error)

    Raises:
        RuntimeError: If error policy determines warning should raise an exception
    """
    # Add warning to collector
    self._warning_collector.add(field, message, severity)

    # Check if we should raise based on error policy
    policy = get_error_policy()
    if policy.should_raise(severity, self._bulk_operation_context):
        # Format message for exception
        if isinstance(message, str):
            error_msg = message
        elif isinstance(message, list):
            error_msg = "; ".join(str(m) for m in message)
        elif isinstance(message, dict):
            error_msg = "; ".join(f"{k}: {v}" for k, v in message.items())
        else:
            error_msg = str(message)

        raise RuntimeError(
            f"{self.__class__.__name__} validation failed on field '{field}': {error_msg}"
        )

set_bulk_context

set_bulk_context(is_bulk=True)

Set whether this model is being processed in a bulk operation context.

Parameters:

Name	Type	Description	Default
is_bulk	bool	True if in bulk operation, False for single operation	True

Note

This affects error handling behavior. In bulk operations, errors are collected as warnings to allow partial success. In single operations, errors may raise exceptions based on the error_mode setting.

Source code in src/pyetm/models/base.py

def set_bulk_context(self, is_bulk: bool = True) -> None:
    """Set whether this model is being processed in a bulk operation context.

    Args:
        is_bulk: True if in bulk operation, False for single operation

    Note:
        This affects error handling behavior. In bulk operations, errors are
        collected as warnings to allow partial success. In single operations,
        errors may raise exceptions based on the error_mode setting.
    """
    self._bulk_operation_context = is_bulk

show_warnings

show_warnings()

Print all warnings to the console.

Source code in src/pyetm/models/base.py

def show_warnings(self) -> None:
    """Print all warnings to the console."""
    self._warning_collector.show_warnings()

auto_show_warnings

auto_show_warnings(context='')

Automatically display warnings if any exist, with contextual information.

Parameters:

Name	Type	Description	Default
context	str	Additional context to display (e.g., "SavedScenario #123")	''

Source code in src/pyetm/models/base.py

def auto_show_warnings(self, context: str = "") -> None:
    """
    Automatically display warnings if any exist, with contextual information.

    Args:
        context: Additional context to display (e.g., "SavedScenario #123")
    """
    if len(self.warnings) > 0:
        if context:
            print(f"\n=== Warnings for {context} ===")
        self.show_warnings()

log_warnings

log_warnings(logger, level='warning', prefix=None)

Log all collected warnings using the provided logger.

Source code in src/pyetm/models/base.py

def log_warnings(
    self, logger: Any, level: str = "warning", prefix: str | None = None
) -> None:
    """
    Log all collected warnings using the provided logger.
    """
    try:
        collector = getattr(self, "warnings", None)
        if collector is None or len(collector) == 0:
            return
        log_fn = getattr(logger, level, getattr(logger, "warning", None))
        if log_fn is None:
            return
        for w in collector:
            field = getattr(w, "field", "")
            msg = getattr(w, "message", str(w))
            if prefix:
                log_fn(f"{prefix} [{field}]: {msg}")
            else:
                log_fn(f"[{field}]: {msg}")
    except Exception:
        pass

from_dataframe classmethod

from_dataframe(df, **kwargs)

Create an instance from a pandas DataFrame.

Source code in src/pyetm/models/base.py

@classmethod
def from_dataframe(cls: Type[T], df: pd.DataFrame, **kwargs: Any) -> T:
    """
    Create an instance from a pandas DataFrame.
    """
    try:
        return cls._from_dataframe(df, **kwargs)
    except Exception as e:
        # Create a fallback instance with warnings
        instance = cls.model_construct()
        instance.add_warning(
            "from_dataframe", f"Failed to create from DataFrame: {e}"
        )
        return instance

to_dataframe

to_dataframe(**kwargs)

Public method that handles common serialization logic and delegates to _to_dataframe().

Returns:

Type	Description
DataFrame	pd.DataFrame: Serialized DataFrame with class name as index level

Source code in src/pyetm/models/base.py

def to_dataframe(self, **kwargs: Any) -> pd.DataFrame:
    """
    Public method that handles common serialization logic and delegates to _to_dataframe().

    Returns:
        pd.DataFrame: Serialized DataFrame with class name as index level
    """
    columns = self._get_serializable_fields()
    kwargs.setdefault("available_columns", columns)

    # Get DataFrame with unified error handling
    try:
        df = self._to_dataframe(**kwargs)
    except Exception as e:
        self.add_warning(
            f"{self.__class__.__name__}._to_dataframe()", f"failed: {e}"
        )
        df = pd.DataFrame()

    return df

Functions