Objects Instances

ObjectInstances(perfdb)

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

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/object_instances.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Class used for handling object instances. Can be accessed via `perfdb.objects.instances`.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.
    """
    super().__init__(perfdb)

    from .object_instance_attributes import ObjectInstanceAttributes

    # * subclasses

    self.attributes = ObjectInstanceAttributes(perfdb)

delete(object_names)

Deletes object instances.

Parameters:

• object_names

  (list[str]) –

  List of object names to delete.

Source code in echo_postgres/object_instances.py

@validate_call
def delete(
    self,
    object_names: list[str],
) -> None:
    """Deletes object instances.

    Parameters
    ----------
    object_names : list[str]
        List of object names to delete.
    """
    query = sql.SQL("DELETE FROM performance.objects WHERE name IN ({names})").format(
        names=sql.SQL(", ").join(map(sql.Literal, object_names)),
    )

    with self._perfdb.conn.reconnect() as conn:
        result = conn.execute(query)

    logger.info(f"Deleted {result.rowcount} object instances")

get(object_names=None, object_models=None, object_types=None, parent_objects=None, object_groups=None, spe_names=None, attributes=None, filter_type='and', get_attributes=False, attribute_names=None, output_type='dict')

Gets all object instances definitions with detailed information.

The most useful keys/columns returned are:

• id
• display_name
• description
• object_type_name
• object_model_name
• parent_object_name
• spe_name

Parameters:

• object_names

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

  List of object names to filter the results. By default None

• object_models

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

  List of object model names to filter the results. By default None

• object_types

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

  List of object type names to filter the results. By default None

• object_groups

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

  List of object group names and their respective types to filter the results. It must be in the format [(group_type, group_name), ...] By default None

• spe_names

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

  List of spe names to filter the results. By default None

• attributes

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

  Dictionary with the attributes 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"

• get_attributes

  (bool, default: False ) –

  If True, will also get the attributes of the object instances. It's highly recommended to specify the attribute names to get to speed up the query.

  By default False

• attribute_names

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

  List of attribute names to get if get_attributes is True. In case get_attributes is False, this parameter is ignored. If None, all attributes are 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, str | int]] –

  In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}

• DataFrame –

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

Source code in echo_postgres/object_instances.py

@validate_call
def get(
    self,
    object_names: list[str] | None = None,
    object_models: list[str] | None = None,
    object_types: list[str] | None = None,
    parent_objects: list[str] | None = None,
    object_groups: list[tuple[str, str]] | None = None,
    spe_names: list[str] | None = None,
    attributes: dict[str, Any] | None = None,
    filter_type: Literal["and", "or"] = "and",
    get_attributes: bool = False,
    attribute_names: list[str] | None = None,
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, str | int]] | DataFrame:
    """Gets all object instances definitions with detailed information.

    The most useful keys/columns returned are:

    - id
    - display_name
    - description
    - object_type_name
    - object_model_name
    - parent_object_name
    - spe_name

    Parameters
    ----------
    object_names : list[str] | None, optional
        List of object names to filter the results. By default None
    object_models : list[str] | None, optional
        List of object model names to filter the results. By default None
    object_types : list[str] | None, optional
        List of object type names to filter the results. By default None
    object_groups : list[tuple[str, str]] | None
        List of object group names and their respective types to filter the results.
        It must be in the format [(group_type, group_name), ...]
        By default None
    spe_names : list[str] | None, optional
        List of spe names to filter the results. By default None
    attributes : dict[str, Any] | None, optional
        Dictionary with the attributes 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"
    get_attributes : bool, optional
        If True, will also get the attributes of the object instances. It's highly recommended to specify the attribute names to get
        to speed up the query.

        By default False
    attribute_names : list[str] | None, optional
        List of attribute names to get if get_attributes is True. In case get_attributes is False, this parameter is ignored. If None, all attributes are 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, str | int]]
        In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
    """
    # checking inputs
    where = self._check_get_args(
        object_names=object_names,
        object_models=object_models,
        object_types=object_types,
        parent_objects=parent_objects,
        object_groups=object_groups,
        spe_names=spe_names,
        attributes=attributes,
        filter_type=filter_type,
    )

    query = [
        sql.SQL("SELECT * FROM performance.v_objects "),
        where,
        sql.SQL(" ORDER BY name"),
    ]
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)
    df = df.set_index("name")

    # getting attributes
    if get_attributes:
        # names of the object instances
        got_object_instances = df.index.tolist()
        attrs: DataFrame = self._perfdb.objects.instances.attributes.get(
            object_names=got_object_instances,
            output_type="DataFrame",
            attribute_names=attribute_names,
            values_only=True,
        )
        if attrs.empty:
            logger.debug("No attributes found")
        else:
            # pivot the attributes
            attrs = attrs.reset_index(drop=False).pivot(index="object_name", columns="attribute_name", values="attribute_value")
            # merging the attributes with the object instances
            df = df.merge(attrs, left_index=True, right_index=True, how="left")

    return df.to_dict(orient="index") if output_type == "dict" else df

get_ids(object_names=None, object_models=None, object_types=None, parent_objects=None, object_groups=None, spe_names=None, attributes=None, filter_type='and')

Gets all object instances and their respective ids.

Parameters:

• object_names

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

  List of object names to filter the results. By default None

• object_models

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

  List of object model names to filter the results. By default None

• object_types

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

  List of object type names to filter the results. By default None

• parent_objects

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

  List of parent object names to filter the results. By default None

• object_groups

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

  List of object group names and their respective types to filter the results. It must be in the format [(group_type, group_name), ...] By default None

• spe_names

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

  List of spe names to filter the results. By default None

• attributes

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

  Dictionary with the attributes 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"].

Returns:

• dict[str, int] –

  Dictionary with all object instances and their respective ids in the format {name: id, ...}.

Source code in echo_postgres/object_instances.py

@validate_call
def get_ids(
    self,
    object_names: list[str] | None = None,
    object_models: list[str] | None = None,
    object_types: list[str] | None = None,
    parent_objects: list[str] | None = None,
    object_groups: list[tuple[str, str]] | None = None,
    spe_names: list[str] | None = None,
    attributes: dict[str, Any] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
    """Gets all object instances and their respective ids.

    Parameters
    ----------
    object_names : list[str] | None, optional
        List of object names to filter the results. By default None
    object_models : list[str] | None, optional
        List of object model names to filter the results. By default None
    object_types : list[str] | None, optional
        List of object type names to filter the results. By default None
    parent_objects : list[str] | None, optional
        List of parent object names to filter the results. By default None
    object_groups : list[tuple[str, str]] | None
        List of object group names and their respective types to filter the results.
        It must be in the format [(group_type, group_name), ...]
        By default None
    spe_names : list[str] | None, optional
        List of spe names to filter the results. By default None
    attributes : dict[str, Any] | None, optional
        Dictionary with the attributes to filter the results. It must be in the format {attribute_name: attribute_value, ...}
        By default None
    filter_type : Literal["and", "or"]
        How to treat multiple filters. Can be one of ["and", "or"].

    Returns
    -------
    dict[str, int]
        Dictionary with all object instances and their respective ids in the format {name: id, ...}.
    """
    # checking inputs
    where = self._check_get_args(
        object_names=object_names,
        object_models=object_models,
        object_types=object_types,
        parent_objects=parent_objects,
        object_groups=object_groups,
        spe_names=spe_names,
        attributes=attributes,
        filter_type=filter_type,
    )

    query = [
        sql.SQL("SELECT name, id FROM performance.v_objects "),
        where,
        sql.SQL(" ORDER BY name"),
    ]
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)

    return df.set_index("name").to_dict()["id"]

insert(object_name, object_model_name, display_name=None, description=None, parent_object_name=None, on_conflict='raise')

Inserts a new object instance.

Parameters:

• object_name

  (str) –

  Name of the object instance.

• object_model_name

  (str) –

  Name of the object model.

• display_name

  (str | None, default: None ) –

  Display name of the object instance. By default None

• description

  (str | None, default: None ) –

  Description of the object instance. By default None

• parent_object_name

  (str | None, default: None ) –

  Name of the parent object. By default None

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

@validate_call
def insert(
    self,
    object_name: str,
    object_model_name: str,
    display_name: str | None = None,
    description: str | None = None,
    parent_object_name: str | None = None,
    on_conflict: Literal["raise", "ignore", "update"] = "raise",
) -> None:
    """Inserts a new object instance.

    Parameters
    ----------
    object_name : str
        Name of the object instance.
    object_model_name : str
        Name of the object model.
    display_name : str | None, optional
        Display name of the object instance. By default None
    description : str | None, optional
        Description of the object instance. By default None
    parent_object_name : str | None, optional
        Name of the parent object. By default None
    on_conflict : Literal["raise", "ignore", "update"], optional
        What to do in case of conflict. Can be one of ["raise", "ignore", "update"].
        By default "raise"
    """
    # getting object model id
    object_models = self._perfdb.objects.models.get_ids(object_models=[object_model_name])
    if not object_models:
        raise ValueError(f"Object model {object_model_name} does not exist")
    object_model_id = object_models[object_model_name]

    # checking if the object instance already exists
    object_instances = self.get_ids(object_names=[object_name])
    object_instance_id = None
    if object_instances:
        if on_conflict == "raise":
            raise ValueError(f"Object instance {object_name} already exists")
        if on_conflict == "ignore":
            return
        if on_conflict == "update":
            object_instance_id = object_instances[object_name]

    # getting parent object id
    parent_object_id = None
    if parent_object_name:
        parent_objects = self.get_ids(object_names=[parent_object_name])
        if not parent_objects:
            raise ValueError(f"Parent object {parent_object_name} does not exist")
        parent_object_id = parent_objects[parent_object_name]

    # building the query
    if object_instance_id:
        query = sql.SQL(
            "UPDATE performance.objects SET object_model_id = {object_model_id}, display_name = {display_name}, description = {description}, parent_object_id = {parent_object_id} WHERE id = {object_instance_id}",
        ).format(
            object_model_id=sql.Literal(object_model_id),
            display_name=sql.Literal(display_name),
            description=sql.Literal(description),
            parent_object_id=sql.Literal(parent_object_id),
            object_instance_id=sql.Literal(object_instance_id),
        )
    else:
        query = sql.SQL(
            "INSERT INTO performance.objects (name, object_model_id, display_name, description, parent_object_id) VALUES ({object_name}, {object_model_id}, {display_name}, {description}, {parent_object_id})",
        ).format(
            object_name=sql.Literal(object_name),
            object_model_id=sql.Literal(object_model_id),
            display_name=sql.Literal(display_name),
            description=sql.Literal(description),
            parent_object_id=sql.Literal(parent_object_id),
        )

    with self._perfdb.conn.reconnect() as conn:
        conn.execute(query)

    logger.info(f"Object instance {object_name} inserted/updated")