utils

ProgressBar

Bases: ProgressType

Progressbar for MinIO function's progress parameter.

Source code in openml/utils.py

class ProgressBar(ProgressType):
    """Progressbar for MinIO function's `progress` parameter."""

    def __init__(self) -> None:
        self._object_name = ""
        self._progress_bar: tqdm | None = None

    def set_meta(self, object_name: str, total_length: int) -> None:
        """Initializes the progress bar.

        Parameters
        ----------
        object_name: str
          Not used.

        total_length: int
          File size of the object in bytes.
        """
        self._object_name = object_name
        self._progress_bar = tqdm(total=total_length, unit_scale=True, unit="B")

    def update(self, length: int) -> None:
        """Updates the progress bar.

        Parameters
        ----------
        length: int
          Number of bytes downloaded since last `update` call.
        """
        if not self._progress_bar:
            raise RuntimeError("Call `set_meta` before calling `update`.")
        self._progress_bar.update(length)
        if self._progress_bar.total <= self._progress_bar.n:
            self._progress_bar.close()

set_meta(object_name, total_length)

Initializes the progress bar.

Parameters:

Name	Type	Description	Default
object_name	str	Not used.	required
total_length	int	File size of the object in bytes.	required

Source code in openml/utils.py

def set_meta(self, object_name: str, total_length: int) -> None:
    """Initializes the progress bar.

    Parameters
    ----------
    object_name: str
      Not used.

    total_length: int
      File size of the object in bytes.
    """
    self._object_name = object_name
    self._progress_bar = tqdm(total=total_length, unit_scale=True, unit="B")

update(length)

Updates the progress bar.

Parameters:

Name	Type	Description	Default
length	int	Number of bytes downloaded since last update call.	required

Source code in openml/utils.py

def update(self, length: int) -> None:
    """Updates the progress bar.

    Parameters
    ----------
    length: int
      Number of bytes downloaded since last `update` call.
    """
    if not self._progress_bar:
        raise RuntimeError("Call `set_meta` before calling `update`.")
    self._progress_bar.update(length)
    if self._progress_bar.total <= self._progress_bar.n:
        self._progress_bar.close()

extract_xml_tags(xml_tag_name, node, *, allow_none=True)

extract_xml_tags(xml_tag_name: str, node: Mapping[str, Any], *, allow_none: Literal[True] = ...) -> Any | None

extract_xml_tags(xml_tag_name: str, node: Mapping[str, Any], *, allow_none: Literal[False]) -> Any

Helper to extract xml tags from xmltodict.

Parameters:

Name	Type	Description	Default
xml_tag_name	str	Name of the xml tag to extract from the node.	required
node	Mapping[str, Any]	Node object returned by xmltodict from which xml_tag_name should be extracted.	required
allow_none	bool	If False, the tag needs to exist in the node. Will raise a ValueError if it does not.	True

Returns:

Type	Description
object

Source code in openml/utils.py

def extract_xml_tags(
    xml_tag_name: str,
    node: Mapping[str, Any],
    *,
    allow_none: bool = True,
) -> Any | None:
    """Helper to extract xml tags from xmltodict.

    Parameters
    ----------
    xml_tag_name : str
        Name of the xml tag to extract from the node.

    node : Mapping[str, Any]
        Node object returned by ``xmltodict`` from which ``xml_tag_name``
        should be extracted.

    allow_none : bool
        If ``False``, the tag needs to exist in the node. Will raise a
        ``ValueError`` if it does not.

    Returns
    -------
    object
    """
    if xml_tag_name in node and node[xml_tag_name] is not None:
        if isinstance(node[xml_tag_name], (dict, str)):
            return [node[xml_tag_name]]
        if isinstance(node[xml_tag_name], list):
            return node[xml_tag_name]

        raise ValueError("Received not string and non list as tag item")

    if allow_none:
        return None

    raise ValueError(f"Could not find tag '{xml_tag_name}' in node '{node!s}'")