Object Instance Attributes

ObjectInstanceAttributeDefinitions(baze)

Class used for handling object instance attribute definitions.

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_names, output_type='dict')

Gets the attribute definitions for specific object instances.

These can be viewed in the portal in Object Types > Object Attribute Definition. Most of the attributes will have the following properties:

• attributeId
• typeId
• categoryId
• displayId
• groupId
• isBuiltIn
• isMandatory
• isVisible
• name
• valueType
• valueTypeArgs

Parameters:

• object_names

  (list[str]) –

  List of object instances 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_instance_attribute_definitions.py

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

    These can be viewed in the portal in `Object Types > Object Attribute Definition`.
    Most of the attributes will have the following properties:

    - attributeId
    - typeId
    - categoryId
    - displayId
    - groupId
    - isBuiltIn
    - isMandatory
    - isVisible
    - name
    - valueType
    - valueTypeArgs

    Parameters
    ----------
    object_names : list[str]
        List of object instances 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
    objects_def = self.baze.objects.instances.get(object_names=object_names)
    object_types = {
        k: {"typeId": v["attributes"]["objectTypeId"], "categoryId": v["attributes"]["objectCategoryId"]}
        for k, v in objects_def.items()
        if k in object_names
    }
    if not object_types:
        raise ValueError("No valid objects were given.")

    attributes_def = []
    for object_name, object_type in object_types.items():
        # getting the data
        endpoint = f"configuration/objecttype/{object_type['typeId']}/category/{object_type['categoryId']}/attributes"
        result = self.baze.conn.get(endpoint)
        self._handle_http_errors(result)
        # converting to dict
        result: dict[str, Any] = result.json()
        # converting to a better format
        # {attribute_key: {name: name, value: value, valueType: valueType}, ...}
        result = {entry["key"]: entry for entry in result}
        # removing key
        for attribute in result.values():
            del attribute["key"]
        # adding to root list
        attributes_def.append({"name": object_name} | result)

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

ObjectInstanceAttributeValues(baze)

Class used for handling object attributes.

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_names, attribute_names=None, output_type='dict')

Gets specific attributes values for specific objects.

This is useful if you only need a specific set of attributes for specific objects, making the method faster than the one available at objects.instances that gets all.

The most useful keys/columns returned are:

• objectId
• attributes (all the requested attributes)

Parameters:

• object_names

  (list[str]) –

  List of object names to get the attributes from.

• attribute_names

  (list[str] | None, default: None ) –

  List of attributes to get. If None, all attributes will be returned, by default None

• 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_instance_attribute_values.py

@validate_call
def get(
    self,
    object_names: list[str],
    attribute_names: list[str] | None = None,
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | DataFrame:
    """Gets specific attributes values for specific objects.

    This is useful if you only need a specific set of attributes for specific objects, making the method faster than the one available at objects.instances that gets all.

    The most useful keys/columns returned are:

    - objectId
    - attributes (all the requested attributes)

    Parameters
    ----------
    object_names : list[str]
        List of object names to get the attributes from.
    attribute_names : list[str] | None, optional
        List of attributes to get. If None, all attributes will be returned, by default None
    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, ...]

    """
    endpoint = "objects/attributes"

    # getting object ids
    object_ids = self.baze.objects.instances.get_ids()
    object_ids = [object_ids[name] for name in object_names if name in object_ids]
    if len(object_ids) != len(object_names):
        raise ValueError(f"Invalid object names: {object_names}")

    # defining payload
    payload = {"ObjectIds": object_names}
    if attribute_names is not None:
        payload |= {"AttributeNames": attribute_names}

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

    # converting to dict
    result: dict[str, Any] = result.json()["data"]
    # converting to list of dicts
    result = list(result.values())
    # casting attributes
    result = self.baze.objects.instances._cast_attributes(result)  # pylint: disable=protected-access # noqa
    # getting only the attributes
    result = [entry["attributes"] for entry in result]
    # keeping only objectKey, objectId and the requested attributes in case they where specified
    if attribute_names is not None:
        final_result = [
            {
                "objectKey": entry["objectKey"],
                "objectId": entry["objectId"],
                **{key: entry[key] for key in attribute_names if key in entry},
            }
            for entry in result
        ]
        result = final_result

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

update(object_name, attributes)

Method used to update attributes for a specific object.

This action can be done in the Object Manager section of the portal.

Parameters:

• object_name

  (str) –

  Name of the desired object

• attributes

  (dict[str, Any]) –

  Dict containing the attributes to update and their values. It must be in the format {attribute: value, ...}

Source code in echo_baze/object_instance_attribute_values.py

@validate_call
def update(self, object_name: str, attributes: dict[str, Any]) -> None:
    """Method used to update attributes for a specific object.

    This action can be done in the Object Manager section of the portal.

    Parameters
    ----------
    object_name : str
        Name of the desired object
    attributes : dict[str, Any]
        Dict containing the attributes to update and their values. It must be in the format {attribute: value, ...}

    """
    # getting objectCategoryId and objectTypeId
    objs_def = self.baze.objects.instances.get(object_names=[object_name])
    if object_name not in objs_def:
        raise ValueError(f"Invalid object name: {object_name}")
    object_id, object_category_id, object_type_id = (
        objs_def[object_name]["objectId"],
        objs_def[object_name]["attributes"]["objectCategoryId"],
        objs_def[object_name]["attributes"]["objectTypeId"],
    )

    # defining payload
    # the PUT API https://bazefield.echoenergia.com.br/BazeField.Services/api/objects needs at least objectCategoryId and objectTypeId to work, the other attributes can be the ones that are being inserted/updated
    payload = {
        "objectId": object_id,
        "attributes": {
            "objectCategoryId": object_category_id,
            "objectTypeId": object_type_id,
        },
    }
    for attribute_key, attribute_value in attributes.items():
        if attribute_key not in objs_def[object_name]["attributes"]:
            raise ValueError(f"Attribute key {attribute_key} is not valid for object {object_name}.")
        payload["attributes"][attribute_key] = attribute_value
    # adding all the other attributes that are not being updated (needed after 10.7 update)
    for attribute_key, attribute_value in objs_def[object_name]["attributes"].items():
        if attribute_key not in payload["attributes"]:
            # if it's a list like ["AA", "BB"] converting it to a string like '["AA","BB"]'
            if isinstance(attribute_value, list):
                if len(attribute_value) > 0 and attribute_value[0] != "":
                    # this is needed for bazefield API to work properly
                    # leaving the space after the comma makes the second owner be ignored
                    set_value = json.dumps(attribute_value).replace(", ", ",")
                else:
                    set_value = "[]"
            else:
                set_value = attribute_value
            payload["attributes"][attribute_key] = set_value

    # updating
    endpoint = "objects"

    result = self.baze.conn.post(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_name, attribute_value in attributes.items():
        if str(attribute_value) != result["data"][object_id]["attributes"][attribute_name]:
            raise ValueError(f"Attribute {attribute_name} was not updated for object {object_name}.")