Skip to content

Allocation Labels

AllocationLabels(baze)

Class used for handling allocation labels.

Source code in echo_baze/baze_root.py
def __init__(self, baze: e_bz.Baze) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    baze : Baze
        Top level object carrying all functionality and the connection handler.

    """
    # check inputs
    if not isinstance(baze, e_bz.Baze):
        raise ValueError(f"baze must be of type Baze, not {type(baze)}")

    self.baze: e_bz.Baze = baze

delete(name)

Deletes an allocation label.

Parameters:

  • name

    (str) –

    Name of the allocation label to delete.

Source code in echo_baze/allocation_labels.py
@validate_call
def delete(self, name: str) -> None:
    """Deletes an allocation label.

    Parameters
    ----------
    name : str
        Name of the allocation label to delete.

    """
    # checking if the allocation label exists
    alloc_ids = self.get_ids()
    if name not in alloc_ids:
        logger.warning(f"Allocation label with name '{name}' does not exist. Skipping deletion.")
        return

    # creating the payload
    payload = {"label": {"name": name, "id": alloc_ids[name], "type": "Allocation"}}

    # deleting
    endpoint = "labels/delete"

    result = self.baze.conn.post(endpoint, json=payload)
    self._handle_http_errors(result)

    # checking if the allocation label was deleted
    result = result.json()
    if not result:
        raise RuntimeError(f"Allocation label with name '{name}' was not deleted")

get_by_allocation_ids(allocation_ids, chunk_size=200)

Gets the labels for a list of allocation IDs.

Requests are split into chunks to avoid URL length limits.

Parameters:

  • allocation_ids

    (list[int]) –

    List of allocation IDs to get labels for.

  • chunk_size

    (int, default: 200 ) –

    Number of allocation IDs sent per request. By default 200.

Returns:

  • dict[int, list[str]]

    Dictionary in the format {allocation_id: [label_name, ...], ...}. Empty list if no labels.

Source code in echo_baze/allocation_labels.py
@validate_call
def get_by_allocation_ids(self, allocation_ids: list[int], chunk_size: int = 200) -> dict[int, list[str]]:
    """Gets the labels for a list of allocation IDs.

    Requests are split into chunks to avoid URL length limits.

    Parameters
    ----------
    allocation_ids : list[int]
        List of allocation IDs to get labels for.
    chunk_size : int, optional
        Number of allocation IDs sent per request. By default 200.

    Returns
    -------
    dict[int, list[str]]
        Dictionary in the format {allocation_id: [label_name, ...], ...}. Empty list if no labels.

    """
    if not allocation_ids:
        return {}
    combined: dict[int, list[str]] = {}
    for i in range(0, len(allocation_ids), chunk_size):
        chunk = allocation_ids[i : i + chunk_size]
        endpoint = f"labels/allocation?allocationIds={','.join(str(a) for a in chunk)}"
        result = self.baze.conn.get(endpoint)
        self._handle_http_errors(result)
        combined.update(self._parse_labels_response(result.json()))
    return combined

get_ids()

Gets a dictionary with all allocation label names and ids.

Returns:

  • dict[str, int]:

    Dictionary with all allocation labels in the format {name: id, ...}

Source code in echo_baze/allocation_labels.py
def get_ids(self) -> dict[str, int]:
    """Gets a dictionary with all allocation label names and ids.

    Returns
    -------
    dict[str, int]:
        Dictionary with all allocation labels in the format {name: id, ...}

    """
    endpoint = "allocationLabels"

    # getting the data
    result = self.baze.conn.get(endpoint)
    self._handle_http_errors(result)

    # converting to dict
    result: list[dict[str, Any]] = result.json()

    return {entry["name"]: entry["id"] for entry in result}

insert(name)

Inserts a new allocation label.

Parameters:

  • name

    (str) –

    Name of the allocation label.

Returns:

  • int

    Id of the newly created allocation label.

Source code in echo_baze/allocation_labels.py
@validate_call
def insert(self, name: str) -> int:
    """Inserts a new allocation label.

    Parameters
    ----------
    name : str
        Name of the allocation label.

    Returns
    -------
    int
        Id of the newly created allocation label.

    """
    # checking if the allocation label already exists
    alloc_ids = self.get_ids()
    if name in alloc_ids:
        logger.warning(f"Allocation label with name '{name}' already exists. Returning its id.")
        return alloc_ids[name]

    # creating the payload
    payload = {"label": {"name": name}}

    # inserting
    endpoint = "allocationlabels/add"

    result = self.baze.conn.post(endpoint, json=payload)
    self._handle_http_errors(result)

    # checking if the allocation label was created
    result = result.json()
    if result["name"] != name:
        raise RuntimeError(f"Allocation label with name '{name}' was not created")

    return result["id"]

update(name, new_name)

Updates an allocation label.

Parameters:

  • name

    (str) –

    Name of the allocation label to update.

  • new_name

    (str) –

    New name of the allocation label.

Source code in echo_baze/allocation_labels.py
@validate_call
def update(self, name: str, new_name: str) -> None:
    """Updates an allocation label.

    Parameters
    ----------
    name : str
        Name of the allocation label to update.
    new_name : str
        New name of the allocation label.

    """
    # checking if the allocation label exists
    alloc_ids = self.get_ids()
    if name not in alloc_ids:
        raise ValueError(f"Allocation label with name '{name}' does not exist")

    # creating the payload
    payload = {"label": {"name": new_name, "id": alloc_ids[name], "edit": True, "type": "Allocation"}}

    # updating
    endpoint = "labels/save"

    result = self.baze.conn.post(endpoint, json=payload)
    self._handle_http_errors(result)

    # checking if the allocation label was updated
    result = result.json()
    if result["name"] != new_name:
        raise RuntimeError(f"Allocation label with name '{name}' was not updated")