Object Instances - Attributes

ObjectInstanceAttributes(perfdb)

Class used for handling object instances attributes. Can be accessed via perfdb.objects.instances.attributes.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.

    """
    self._perfdb: e_pg.PerfDB = perfdb

delete(object_name, attribute_name)

Deletes an attribute value.

Parameters:

• object_name

  (str) –

  Name of the object instance to delete the attribute value from.

• attribute_name

  (str) –

  Name of the attribute to delete the value from.

Source code in echo_postgres/object_instance_attributes.py

@validate_call
def delete(
    self,
    object_name: str,
    attribute_name: str,
) -> None:
    """Deletes an attribute value.

    Parameters
    ----------
    object_name : str
        Name of the object instance to delete the attribute value from.
    attribute_name : str
        Name of the attribute to delete the value from.
    """
    # building the query
    query = [
        sql.SQL(
            "DELETE FROM performance.object_attributes "
            "WHERE object_id = (SELECT id FROM performance.objects WHERE name = {object_name}) "
            "AND attribute_id = (SELECT id FROM performance.attributes_def WHERE name = {attribute_name}) ",
        ).format(
            object_name=sql.Literal(object_name),
            attribute_name=sql.Literal(attribute_name),
        ),
    ]
    # executing the query
    with self._perfdb.conn.reconnect() as conn:
        # deleting
        result = conn.execute(sql.Composed(query))

    logger.debug(f"Deleted {result.rowcount} rows from performance.object_attributes")

get(object_names=None, attribute_names=None, attribute_values=None, filter_type='and', output_type='dict', values_only=False)

Method to get the attributes of the given object instances.

The most useful keys/columns returned are:

• object_id
• object_name
• object_model_name
• attribute_id
• attribute_name
• attribute_display_name
• attribute_value
• data_type_name

Parameters:

• object_names

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

  List of object instances to get the attributes from. If set to None will get from all. By default None

• attribute_names

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

  List of attribute names to filter the results. If set to None will get all. By default None

• attribute_values

  (dict[str, Any] | None, default: None ) –

  Dictionary of attribute names and values to filter the results. It must be in the format {attribute_name: attribute_value, ...}. By default None

• filter_type

  (Literal['and', 'or'], default: 'and' ) –

  How to treat multiple filters. Can be one of ["and", "or"]. By default "and"

• output_type

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

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

• values_only

  (bool, default: False ) –

  If set to True, will only return the values of the attributes, skipping display_name, id, etc.

Returns:

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

  In case output_type is "dict", returns a dictionary in the format {object_name: {attribute_name: {attribute: value, ...}, ...}, ...} If values_only is set to True, returns a dictionary in the format {object_name: {attribute_name: value, ...}, ...}

• DataFrame –

  In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[object_name, attribute_name], columns = [attribute, ...] If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[object_name, attribute_name], columns = ["attribute_value"]

Source code in echo_postgres/object_instance_attributes.py

@validate_call
def get(
    self,
    object_names: list[str] | None = None,
    attribute_names: list[str] | None = None,
    attribute_values: dict[str, Any] | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame"] = "dict",
    values_only: bool = False,
) -> dict[str, dict[str, Any | dict[str, Any]]] | DataFrame:
    """Method to get the attributes of the given object instances.

    The most useful keys/columns returned are:

    - object_id
    - object_name
    - object_model_name
    - attribute_id
    - attribute_name
    - attribute_display_name
    - attribute_value
    - data_type_name

    Parameters
    ----------
    object_names : list[str] | None, optional
        List of object instances to get the attributes from. If set to None will get from all. By default None
    attribute_names : list[str] | None, optional
        List of attribute names to filter the results. If set to None will get all. By default None
    attribute_values : dict[str, Any] | None, optional
        Dictionary of attribute names and values to filter the results. It must be in the format {attribute_name: attribute_value, ...}.
        By default None
    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"].
        By default "and"
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"
    values_only : bool, optional
        If set to True, will only return the values of the attributes, skipping display_name, id, etc.

    Returns
    -------
    dict[str, dict[str, Any | dict[str, Any]]]
        In case output_type is "dict", returns a dictionary in the format {object_name: {attribute_name: {attribute: value, ...}, ...}, ...}
        If values_only is set to True, returns a dictionary in the format {object_name: {attribute_name: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[object_name, attribute_name], columns = [attribute, ...]
        If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[object_name, attribute_name], columns = ["attribute_value"]
    """
    # checking if all object instances are valid
    if object_names:
        existing_instances = self._perfdb.objects.instances.get_ids(object_names=object_names)
        if missing_instances := set(object_names) - set(
            existing_instances.keys(),
        ):
            raise ValueError(f"The following object instances do not exist: {missing_instances}")

    # checking if all attribute names are valid and also getting its ids
    if attribute_names or attribute_values:
        wanted_attrs = set()
        if attribute_names:
            wanted_attrs = set(attribute_names)
        if attribute_values:
            wanted_attrs = wanted_attrs.union(set(attribute_values.keys()))
        # removing columns from v_objects from wanted_attrs
        wanted_attrs = wanted_attrs.difference(
            {
                "id",
                "name",
                "display_name",
                "description",
                "object_model_id",
                "object_model_name",
                "object_type_id",
                "object_type_name",
                "parent_object_id",
                "parent_object_name",
                "spe_id",
                "spe_name",
            },
        )

        wanted_attrs = list(wanted_attrs)
        existing_attributes = self._perfdb.attributes.get_ids(attribute_names=wanted_attrs)
        if missing_attributes := set(wanted_attrs) - set(
            existing_attributes.keys(),
        ):
            raise ValueError(f"The following attribute names do not exist: {missing_attributes}")

    # building the query
    query = [
        sql.SQL("SELECT {values} FROM performance.v_object_attributes").format(
            values=sql.SQL(
                "object_name, attribute_name, attribute_value, data_type_name",
            )
            if values_only
            else sql.SQL("*"),
        ),
    ]
    where = []
    if object_names:
        where.append(
            sql.SQL(" object_id IN ({object_instances}) ").format(
                object_instances=sql.SQL(",").join(sql.Literal(oid) for oid in existing_instances.values()),
            ),
        )
    if attribute_names:
        where.append(
            sql.SQL(" attribute_id IN ({attribute_ids}) ").format(
                attribute_ids=sql.SQL(",").join(sql.Literal(aid) for aid in existing_attributes.values()),
            ),
        )
    if attribute_values:
        attr_vals_query = [
            sql.SQL(
                " (attribute_id = {attribute_id} AND attribute_value = {attribute_value}::TEXT::JSONB) ",
            ).format(
                attribute_id=sql.Literal(existing_attributes[an]),
                attribute_value=sql.Literal(f'"{av}"') if isinstance(av, str) else sql.Literal(av),
            )
            for an, av in attribute_values.items()
        ]
        attr_vals_query = sql.SQL(f" {filter_type.upper()} ").join(attr_vals_query)
        where.append(attr_vals_query)
    if where:
        where = sql.SQL(f" {filter_type.upper()} ").join(where)
        query.append(sql.SQL(" WHERE "))
        query.append(where)
    query.append(sql.SQL(" ORDER BY object_name, attribute_name"))
    query = sql.Composed(query)

    # executing the query
    with self._perfdb.conn.reconnect() as conn:
        # setting attribute_value as object to avoid casting json column as string
        df = conn.read_to_pandas(query, post_convert="pyarrow")

    # casting the attribute values
    df = cast_attributes(df=df, index_cols=["object_name"])

    df = df.set_index(["object_name", "attribute_name"])

    # returning the result
    if output_type == "dict":
        # dropping unwanted columns
        if values_only:
            df = df["attribute_value"]
            output = df.to_dict()
        else:
            output = df[["attribute_id", "attribute_value", "data_type_id", "data_type_name", "modified_date"]].to_dict(orient="index")
        # converting dict where the keys are tuples {(key1, key2): value}, to a dict where the keys are strings like {key1: {key2: value}}
        new_output = {}
        for (om, an), values in output.items():
            if om not in new_output:
                new_output[om] = {}
            new_output[om][an] = values
        return new_output
    if output_type == "DataFrame" and values_only:
        df = df[["attribute_value"]].copy()

    return df

insert(object_name, attribute_name, attribute_value, on_conflict='raise')

Inserts a new attribute value.

Parameters:

• object_name

  (str) –

  Name of the object instance to insert the attribute value to.

• attribute_name

  (str) –

  Name of the attribute to insert the value to.

• attribute_value

  (Any) –

  Value of the attribute.

• on_conflict

  (Literal['raise', 'ignore', 'update'], default: 'raise' ) –

  What to do in case of conflict. Can be one of ["raise", "ignore", "update"]. By default "raise"

Source code in echo_postgres/object_instance_attributes.py

@validate_call
def insert(
    self,
    object_name: str,
    attribute_name: str,
    attribute_value: Any,
    on_conflict: Literal["raise", "ignore", "update"] = "raise",
) -> None:
    """Inserts a new attribute value.

    Parameters
    ----------
    object_name : str
        Name of the object instance to insert the attribute value to.
    attribute_name : str
        Name of the attribute to insert the value to.
    attribute_value : Any
        Value of the attribute.
    on_conflict : Literal["raise", "ignore", "update"], optional
        What to do in case of conflict. Can be one of ["raise", "ignore", "update"].
        By default "raise"
    """
    # checking if object instance exists
    existing_instances = self._perfdb.objects.instances.get_ids(object_names=[object_name])
    if object_name not in existing_instances:
        raise ValueError(f"The object instance {object_name} does not exist")

    # checking and casting the attribute value
    insert_attribute_value, attribute_id = check_attribute_dtype(
        attribute_name=attribute_name,
        attribute_value=attribute_value,
        perfdb=self._perfdb,
    )

    # building the query
    query = [
        sql.SQL(
            "INSERT INTO performance.object_attributes (object_id, attribute_id, value) "
            "VALUES ({object_id}, {attribute_id}, {attribute_value}) ",
        ).format(
            object_id=sql.Literal(existing_instances[object_name]),
            attribute_id=sql.Literal(attribute_id),
            attribute_value=sql.Literal(insert_attribute_value),
        ),
    ]
    match on_conflict:
        case "raise":
            # doing nothing will raise conflicts as expected
            pass
        case "ignore":
            query.append(sql.SQL(" ON CONFLICT DO NOTHING "))
        case "update":
            query.append(
                sql.SQL(
                    " ON CONFLICT (object_id, attribute_id) DO UPDATE SET value = EXCLUDED.value ",
                ),
            )

    # executing the query
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(sql.Composed(query))

    logger.debug(f"Attribute '{attribute_name}' inserted to object instance '{object_name}'")

update(object_name, attribute_name, attribute_value)

Updates an attribute value.

Parameters:

• object_name

  (str) –

  Name of the object instance to update the attribute value from.

• attribute_name

  (str) –

  Name of the attribute to update the value from.

• attribute_value

  (Any) –

  New value of the attribute.

Source code in echo_postgres/object_instance_attributes.py

@validate_call
def update(
    self,
    object_name: str,
    attribute_name: str,
    attribute_value: Any,
) -> None:
    """Updates an attribute value.

    Parameters
    ----------
    object_name : str
        Name of the object instance to update the attribute value from.
    attribute_name : str
        Name of the attribute to update the value from.
    attribute_value : Any
        New value of the attribute.
    """
    # checking if object instance exists
    existing_instances = self._perfdb.objects.instances.get_ids(object_names=[object_name])
    if object_name not in existing_instances:
        raise ValueError(f"The object instance {object_name} does not exist")

    # checking and casting the attribute value
    insert_attribute_value, attribute_id = check_attribute_dtype(
        attribute_name=attribute_name,
        attribute_value=attribute_value,
        perfdb=self._perfdb,
    )

    # building the query
    query = [
        sql.SQL(
            "UPDATE performance.object_attributes "
            "SET value = {attribute_value} "
            "WHERE object_id = {object_id} "
            "AND attribute_id = {attribute_id} ",
        ).format(
            object_id=sql.Literal(existing_instances[object_name]),
            attribute_id=sql.Literal(attribute_id),
            attribute_value=sql.Literal(insert_attribute_value),
        ),
    ]
    # executing the query
    with self._perfdb.conn.reconnect() as conn:
        conn.execute(sql.Composed(query))

    logger.debug(f"Attribute '{attribute_name}' updated in object instance '{object_name}'")