Object Type Attributes

ObjectTypeAttributeValues(baze)

Class used for handling object attribute values.

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

get(object_types, output_type='dict')

Gets the attribute values for specific object types.

These can be viewed in the portal in Object Types > Properties.

Parameters:

• object_types

  (list[str]) –

  List of object types to get the attributes from.

• output_type

  (Literal['dict', 'DataFrame'], default: 'dict' ) –

  Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"

Returns:

• dict[str, dict[str, Any]] –

  In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}

• DataFrame –

  In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]

Source code in echo_baze/object_type_attribute_values.py

@validate_call
def get(self, object_types: list[str], output_type: Literal["dict", "DataFrame"] = "dict") -> dict[str, dict[str, Any]] | DataFrame:
    """Gets the attribute values for specific object types.

    These can be viewed in the portal in Object Types > Properties.

    Parameters
    ----------
    object_types : list[str]
        List of object types to get the attributes from.
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, Any]]
        In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}
    DataFrame
        In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]

    """
    # getting object type classId
    object_type_class_ids = self.baze.objects.types.get()
    object_type_class_ids = {k: v["type"]["classId"] for k, v in object_type_class_ids.items() if k in object_types}
    if not object_type_class_ids:
        raise ValueError("No valid object types were given.")

    results = []
    for object_type_name, class_id in object_type_class_ids.items():
        endpoint = f"objectTypes/objectTypeByClassId/{class_id}"
        # getting the data
        result = self.baze.conn.get(endpoint)
        self._handle_http_errors(result)
        # converting to dict
        result: dict[str, Any] = result.json()
        results.append({"typeName": object_type_name, "attributes": result["attributes"]})

    # converting to a better format
    # {type_name: {attribute_key: {name: name, value: value, valueType: valueType}, ...}, ...}
    for i in range(len(results)):  # pylint: disable=consider-using-enumerate
        results[i]["attributes"] = {entry["key"]: entry for entry in results[i]["attributes"]}
        # removing key
        for attribute in results[i]["attributes"].values():
            del attribute["key"]

        # adding to root dict
        results[i] = results[i] | results[i]["attributes"]
        del results[i]["attributes"]

    # returning on the desired format
    match output_type:
        case "dict":
            # defining the name as the key for the dict
            result_dict = {entry["typeName"]: entry for entry in results}
            # removing the name from the dict values
            for entry in result_dict.values():
                del entry["typeName"]
            return result_dict
        case "DataFrame":
            df = json_normalize(results, max_level=1)
            df = df.convert_dtypes(dtype_backend="pyarrow")
            df = df.set_index("typeName")
            return df
        case _:
            raise ValueError(f"Invalid output_type: {output_type}")

update(object_type, attributes)

Updates attribute values for a specific object type.

This action can be done in the portal in Object Types > Properties.

Parameters:

• object_type

  (str) –

  Name of the desired object type.

• attributes

  (dict[str, Any]) –

  Dict with the attributes to be updated. It must be in the format {key: value, ...}

Source code in echo_baze/object_type_attribute_values.py

@validate_call
def update(self, object_type: str, attributes: dict[str, Any]) -> None:
    """Updates attribute values for a specific object type.

    This action can be done in the portal in Object Types > Properties.

    Parameters
    ----------
    object_type : str
        Name of the desired object type.
    attributes : dict[str, Any]
        Dict with the attributes to be updated. It must be in the format {key: value, ...}

    """
    # getting object type classId
    object_type_class_id = self.baze.objects.types.get()
    if object_type not in object_type_class_id:
        raise ValueError(f"Wanted object type {object_type} is not valid.")
    object_type_class_id = object_type_class_id[object_type]["type"]["classId"]

    # getting type definition
    endpoint = f"objectTypes/objectTypeByClassId/{object_type_class_id}"
    # getting the data
    result = self.baze.conn.get(endpoint)
    self._handle_http_errors(result)
    # converting to dict
    result: dict[str, Any] = result.json()

    # changing attributes
    for attribute_key, attribute_value in attributes.items():
        for i in range(len(result["attributes"])):
            if result["attributes"][i]["key"] == attribute_key:
                result["attributes"][i]["value"] = attribute_value
                break
        else:
            raise ValueError(f"Attribute key {attribute_key} is not valid for object type {object_type}.")

    # updating
    endpoint = "objectTypes/updateObjectType"

    payload = deepcopy(result)

    result = self.baze.conn.put(endpoint, json=payload)
    self._handle_http_errors(result)
    # converting to dict
    result: dict[str, Any] = result.json()

    # checking if the update was successful
    for attribute_value in result["attributes"]:
        for updated_attribute in payload["attributes"]:
            if updated_attribute["key"] == attribute_value["key"]:
                if updated_attribute["value"] != attribute_value["value"]:
                    raise ValueError(f"Updating attribute {attribute_value['key']} failed.")
                break