sessions

sessions

Session management utilities.

Classes

Sessions

Sessions(**data)

Bases: Base

A collection of Scenario objects

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

load_all classmethod

load_all(page=None, per_page=None, client=None)

Load scenarios belonging to the authenticated user.

Fetches ETEngine sessions for the authenticated user. By default, automatically fetches all pages. If page parameter is provided, fetches only that specific page.

Parameters:

Name	Type	Description	Default
page	Optional[int]	Optional page number (1-indexed). If provided, fetches only that page. If None, fetches all pages automatically.	None
per_page	Optional[int]	Optional number of results per page	None
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None

Returns:

Type	Description
'Sessions'	Sessions collection containing user's sessions

Raises:

Type	Description
ValueError	If authentication fails or API request fails

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

@classmethod
def load_all(
    cls,
    page: Optional[int] = None,
    per_page: Optional[int] = None,
    client: Optional[BaseClient] = None,
) -> "Sessions":
    """Load scenarios belonging to the authenticated user.

    Fetches ETEngine sessions for the authenticated user. By default, automatically
    fetches all pages. If page parameter is provided, fetches only that specific page.

    Args:
        page: Optional page number (1-indexed). If provided, fetches only that page.
              If None, fetches all pages automatically.
        per_page: Optional number of results per page
        client: Optional BaseClient instance for API communication

    Returns:
        Sessions collection containing user's sessions

    Raises:
        ValueError: If authentication fails or API request fails
    """
    from pyetm.services.scenario_runners.fetch_user_scenarios import (
        FetchUserScenariosRunner,
    )

    if client is None:
        from pyetm.clients import BaseClient

        client = BaseClient()

    result = FetchUserScenariosRunner.run(
        client=client,
        page=page,
        per_page=per_page,
    )

    if not result.success:
        raise ValueError(
            f"Failed to fetch user scenarios: {'; '.join(result.errors)}"
        )

    if result.data is None:
        raise ValueError("No data returned from API")

    # Use model_validate to avoid N+1 API calls
    scenarios = [Session.model_validate(data) for data in result.data]
    return cls(items=scenarios)

load_many classmethod

load_many(scenario_ids, client=None)

Load multiple Session objects by their ETEngine session IDs.

This is a bulk operation - individual failures are collected as warnings to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

Parameters:

Name	Type	Description	Default
scenario_ids	Iterable[int]	Iterable of ETEngine session IDs to load	required
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None

Returns:

Type	Description
'Sessions'	Sessions collection containing the loaded Session objects.
'Sessions'	Warnings from failures are displayed automatically.

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

@classmethod
def load_many(
    cls, scenario_ids: Iterable[int], client: Optional[BaseClient] = None
) -> "Sessions":
    """Load multiple Session objects by their ETEngine session IDs.

    This is a bulk operation - individual failures are collected as warnings
    to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

    Args:
        scenario_ids: Iterable of ETEngine session IDs to load
        client: Optional BaseClient instance for API communication

    Returns:
        Sessions collection containing the loaded Session objects.
        Warnings from failures are displayed automatically.
    """
    sessions = cls(items=[])
    sessions.set_bulk_context(True)

    for sid in scenario_ids:
        try:
            session = Session.load(sid, client=client)
            session.set_bulk_context(True)
            sessions.items.append(session)
        except ScenarioError as e:
            sessions.add_warning(
                "load_many",
                f"Could not load scenario {sid}: {e}",
                severity="error",
            )

    # Auto-display warnings if any
    if len(sessions.warnings) > 0:
        sessions.show_warnings()

    return sessions

create_many classmethod

create_many(scenario_params, area_code=None, end_year=None)

Create multiple Session objects from parameter dicts.

This is a bulk operation - individual failures are collected as warnings to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

Parameters:

Name	Type	Description	Default
scenario_params	Iterable[dict[str, Any]]	Iterable of parameter dicts containing area_code, end_year, etc.	required
area_code	str | None	Default area_code for all sessions (if not in params)	None
end_year	int | None	Default end_year for all sessions (if not in params)	None

Returns:

Type	Description
'Sessions'	Sessions collection containing created Session objects.
'Sessions'	Warnings from failures are displayed automatically.

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

@classmethod
def create_many(
    cls,
    scenario_params: Iterable[dict[str, Any]],
    area_code: str | None = None,
    end_year: int | None = None,
) -> "Sessions":
    """Create multiple Session objects from parameter dicts.

    This is a bulk operation - individual failures are collected as warnings
    to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

    Args:
        scenario_params: Iterable of parameter dicts containing area_code, end_year, etc.
        area_code: Default area_code for all sessions (if not in params)
        end_year: Default end_year for all sessions (if not in params)

    Returns:
        Sessions collection containing created Session objects.
        Warnings from failures are displayed automatically.
    """
    sessions = cls(items=[])
    sessions.set_bulk_context(True)

    created_sessions = []
    for params in scenario_params:
        area = params.get("area_code") or area_code
        year = params.get("end_year") or end_year
        if area is None or year is None:
            sessions.add_warning(
                "create_many",
                f"Could not create scenario with {params}: Missing area_code or end_year. Provide them in each dict or as defaults.",
                severity="warning",
            )
            continue
        try:
            extra = {
                k: v
                for k, v in params.items()
                if k not in ("area_code", "end_year")
            }
            session = Session.new(area, year, **extra)
            session.set_bulk_context(True)
            created_sessions.append(session)
        except (ScenarioError, ValueError) as e:
            sessions.add_warning(
                "create_many",
                f"Could not create scenario with {params}: {e}",
                severity="error",
            )

    # Add created sessions to collection
    sessions.items = created_sessions

    # Merge warnings from individual sessions
    for session in created_sessions:
        sessions._merge_submodel_warnings(session)

    # Auto-display warnings if any
    if len(sessions.warnings) > 0 or any(len(s.warnings) > 0 for s in created_sessions):
        print("\n=== Batch Creation Summary ===")
        sessions.show_warnings()

    return sessions

to_excel

to_excel(path, **export_options)

Export all scenarios to Excel.

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

def to_excel(self, path: PathLike[str] | str, **export_options: Any) -> None:
    """Export all scenarios to Excel."""
    from pyetm.models.scenario_packer import ScenarioPacker

    if not self.items:
        raise ValueError("No scenarios to export")

    packer = ScenarioPacker()
    packer.add(*self.items)
    packer.to_excel(str(Path(path).expanduser().resolve()), **export_options)

from_excel classmethod

from_excel(xlsx_path)

Import scenarios (Sessions) from Excel file.

Only loads scenarios where the 'session' column is True. SavedScenarios (session=False or missing) are ignored.

Parameters:

Name	Type	Description	Default
xlsx_path	PathLike[str] | str	Path to Excel file	required

Returns:

Type	Description
'Sessions'	Scenarios collection containing only Session instances

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

@classmethod
def from_excel(cls, xlsx_path: PathLike[str] | str) -> "Sessions":
    """
    Import scenarios (Sessions) from Excel file.

    Only loads scenarios where the 'session' column is True.
    SavedScenarios (session=False or missing) are ignored.

    Args:
        xlsx_path: Path to Excel file

    Returns:
        Scenarios collection containing only Session instances
    """
    from pyetm.models.scenario_packer import ScenarioPacker
    from pyetm.models.scenario import Scenario

    resolved_path = Path(xlsx_path).expanduser().resolve()
    packer = ScenarioPacker.from_excel(str(resolved_path))
    all_scenarios = list(packer._scenarios())
    sessions = [s for s in all_scenarios if not isinstance(s, Scenario)]

    if not sessions:
        print(f"No Sessions found in Excel file: {resolved_path}")

    sessions.sort(key=lambda s: s.id if hasattr(s, "id") else 0)
    return cls(items=sessions)