sortables

sortables

Sortable items management for scenarios.

Classes

SortableError

Bases: Exception

Base sortable error

Sortable

Sortable(**data)

Bases: Base

Represents one sortable order. - If payload is a flat list, yields one Sortable. - If payload is a dict (heat_network), yields one Sortable per subtype.

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")

Functions

name

name()

Returns the display name

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

def name(self) -> str:
    """
    Returns the display name
    """
    if self.subtype:
        return f"{self.type}_{self.subtype}"
    else:
        return self.type

is_valid_update

is_valid_update(new_order)

Returns a WarningCollector with validation warnings without updating the current object

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

def is_valid_update(self, new_order: list[Any]) -> WarningCollector:
    """
    Returns a WarningCollector with validation warnings without updating the current object
    """
    warnings = WarningCollector()

    new_obj_dict = self.model_dump()
    new_obj_dict["order"] = new_order

    try:
        warnings_obj = self.__class__(**new_obj_dict)
        if isinstance(warnings_obj.warnings, WarningCollector):
            warnings = warnings_obj.warnings
    except Exception:
        pass

    # Additional validation: check against cached valid items
    if self.valid_items is not None:
        unknown_items = set(new_order) - self.valid_items
        if unknown_items:
            warnings.add(
                "order",
                f"Unknown items not in current sortable: {sorted(unknown_items)}. "
                "Fetch sortables first to see valid items."
            )

    return warnings

validate_type classmethod

validate_type(value)

Validate that type is a non-empty string

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

@field_validator("type")
@classmethod
def validate_type(cls, value: str) -> str:
    """Validate that type is a non-empty string"""
    if not isinstance(value, str) or not value.strip():
        raise ValueError("Type must be a non-empty string")
    return value.strip()

validate_subtype classmethod

validate_subtype(value)

Validate subtype if provided

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

@field_validator("subtype")
@classmethod
def validate_subtype(cls, value: Optional[str]) -> Optional[str]:
    """Validate subtype if provided"""
    if value is not None:
        if not isinstance(value, str) or not value.strip():
            raise ValueError("Subtype must be a non-empty string or None")
        return value.strip()
    return value

validate_order classmethod

validate_order(value)

Validate that order is a list and check for duplicates

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

@field_validator("order")
@classmethod
def validate_order(cls, value: list[Any]) -> list[Any]:
    """Validate that order is a list and check for duplicates"""
    if not isinstance(value, list):
        raise ValueError("Order must be a list")

    # Check for duplicates
    seen = set()
    duplicates = []
    for item in value:
        if item in seen:
            duplicates.append(item)
        seen.add(item)

    if duplicates:
        raise ValueError(f"Order contains duplicate items: {duplicates}")

    return value

validate_sortable_consistency

validate_sortable_consistency()

Additional validation for the entire sortable

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

@model_validator(mode="after")
def validate_sortable_consistency(self) -> "Sortable":
    """Additional validation for the entire sortable"""
    if self.type == "heat_network" and self.subtype is None:
        raise ValueError("heat_network type requires a subtype")

    if len(self.order) > 17:
        raise ValueError("Order cannot contain more than 17 items")

    return self

from_json classmethod

from_json(data)

:param data: (sortable_type, payload) - payload list → yield Sortable(type, order) - payload dict → yield Sortable(type, subtype, order) for each subtype

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

@classmethod
def from_json(
    cls, data: Tuple[str, Union[list[Any], Dict[str, list[Any]]]]
) -> Iterator[Sortable]:
    """
    :param data: (sortable_type, payload)
       - payload list → yield Sortable(type, order)
       - payload dict → yield Sortable(type, subtype, order) for each subtype
    """
    sort_type, payload = data

    if isinstance(payload, list):
        # Extract valid items from the order and cache them
        valid_items = set(payload) if payload else None
        sortable = cls(type=sort_type, order=payload, valid_items=valid_items)
        yield sortable

    elif isinstance(payload, dict):
        for sub, order in payload.items():
            # Extract valid items from the order and cache them
            valid_items = set(order) if order else None
            sortable = cls(type=sort_type, subtype=sub, order=order, valid_items=valid_items)
            yield sortable

    else:
        # Create basic sortable with warning for unexpected payload
        sortable = cls(type=sort_type, order=[])
        sortable.add_warning("payload", f"Unexpected payload for '{sort_type}': {payload!r}")
        yield sortable

Sortables

Sortables(**data)

Bases: Base

A flat collection of Sortable instances, regardless of whether the source JSON was nested.

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")

Functions

names

names()

Get all sortable names (including subtype suffixes)

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

def names(self) -> List[str]:
    """Get all sortable names (including subtype suffixes)"""
    return [s.name() for s in self.sortables]

is_valid_update

is_valid_update(updates)

Returns a dict mapping sortable names to their WarningCollectors when errors were found

:param updates: Dict mapping sortable names to new orders :return: Dict mapping sortable names to WarningCollectors

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

def is_valid_update(self, updates: Dict[str, list[Any]]) -> Dict[str, WarningCollector]:
    """
    Returns a dict mapping sortable names to their WarningCollectors when errors were found

    :param updates: Dict mapping sortable names to new orders
    :return: Dict mapping sortable names to WarningCollectors
    """
    warnings = {}

    # Check each sortable that has an update
    sortable_by_name = {s.name(): s for s in self.sortables}

    for name, new_order in updates.items():
        if name in sortable_by_name:
            sortable = sortable_by_name[name]
            sortable_warnings = sortable.is_valid_update(new_order)
            if len(sortable_warnings) > 0:
                warnings[name] = sortable_warnings
        else:
            warnings[name] = WarningCollector.with_warning(name, "Sortable does not exist")

    # Check for non-existent sortables
    non_existent_names = set(updates.keys()) - set(self.names())
    for name in non_existent_names:
        if name not in warnings:  # Don't overwrite existing warnings
            warnings[name] = WarningCollector.with_warning(name, "Sortable does not exist")

    return warnings

update

update(updates)

Update the orders of specified sortables with validation and warning display.

Invalid orders are rejected (not applied) to maintain data integrity. Warnings are automatically displayed for invalid orders and non-existent sortables. Warnings from previous updates are cleared to show only current operation issues.

Parameters:

Name	Type	Description	Default
updates	Dict[str, list[Any]]	Dictionary mapping sortable names to new orders	required

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

def update(self, updates: Dict[str, list[Any]]) -> None:
    """
    Update the orders of specified sortables with validation and warning display.

    Invalid orders are rejected (not applied) to maintain data integrity.
    Warnings are automatically displayed for invalid orders and non-existent sortables.
    Warnings from previous updates are cleared to show only current operation issues.

    Args:
        updates: Dictionary mapping sortable names to new orders
    """
    # Auto-clear stale warnings from previous updates
    self.warnings.clear()

    sortable_by_name = {s.name(): s for s in self.sortables}

    # Check for non-existent sortable names
    for name in updates.keys():
        if name not in sortable_by_name:
            self.add_warning(name, f"Sortable '{name}' does not exist")

    # Apply updates (Base.__setattr__ will validate and add warnings)
    for name, new_order in updates.items():
        if name in sortable_by_name:
            # Validate first (including unknown items)
            sortable_warnings = sortable_by_name[name].is_valid_update(new_order)
            if len(sortable_warnings) > 0:
                for warning in sortable_warnings:
                    self.add_warning(name, warning.message)

            # Apply update
            sortable_by_name[name].order = new_order
            # Collect any additional warnings from Pydantic validation
            if sortable_by_name[name].warnings.has_warnings("order"):
                for warning in sortable_by_name[name].warnings.get_by_field("order"):
                    # Avoid duplicating warnings we already added
                    if not any(w.message == warning.message for w in sortable_warnings):
                        self.add_warning(name, warning.message)

    # Auto-display warnings if any exist
    if len(self.warnings) > 0:
        self.auto_show_warnings()

validate_sortables_list classmethod

validate_sortables_list(value)

Validate the list of sortables

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

@field_validator("sortables")
@classmethod
def validate_sortables_list(cls, value: List[Sortable]) -> List[Sortable]:
    """Validate the list of sortables"""
    if not isinstance(value, list):
        raise ValueError("Sortables must be a list")

    # Check for duplicate names
    names = [s.name() for s in value if isinstance(s, Sortable)]
    duplicates = []
    seen = set()
    for name in names:
        if name in seen:
            duplicates.append(name)
        seen.add(name)

    if duplicates:
        raise ValueError(f"Duplicate sortable names found: {duplicates}")

    return value

validate_sortables_consistency

validate_sortables_consistency()

Additional validation for the entire sortables collection

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

@model_validator(mode="after")
def validate_sortables_consistency(self) -> "Sortables":
    """Additional validation for the entire sortables collection"""
    heat_network_types = [s for s in self.sortables if s.type == "heat_network"]
    if len(heat_network_types) > 0:
        # All heat_network sortables should have subtypes
        without_subtypes = [s for s in heat_network_types if s.subtype is None]
        if without_subtypes:
            raise ValueError("All heat_network sortables must have subtypes")

    return self

from_json classmethod

from_json(data)

:param data: the raw JSON dict from GET /api/v3/scenarios/:id/user_sortables

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

@classmethod
def from_json(cls, data: Dict[str, Any]) -> "Sortables":
    """
    :param data: the raw JSON dict from
                 GET /api/v3/scenarios/:id/user_sortables
    """
    items: List[Sortable] = []
    for pair in data.items():
        items.extend(Sortable.from_json(pair))

    # Use Base class constructor that handles validation gracefully
    collection = cls(sortables=items)

    collection._merge_submodel_warnings(*items, key_attr="type")

    return collection

as_dict

as_dict()

Return a dict mimicking the index endpoint.

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

def as_dict(self) -> Dict[str, Any]:
    """
    Return a dict mimicking the index endpoint.
    """
    result: Dict[str, Any] = {}
    for s in self.sortables:
        if s.subtype:
            result.setdefault(s.type, {})[s.subtype] = s.order
        else:
            result[s.type] = s.order
    return result