Label Values¶
LabelValues(perfdb)
¶
Class used for handling label values. Can be accessed via perfdb.kpis.labels.values.
Parameters:
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(labels, label_type_name)
¶
Deletes a label value.
Parameters:
Source code in echo_postgres/label_values.py
@validate_call
def delete(
self,
labels: list[str],
label_type_name: str,
) -> None:
"""Deletes a label value.
Parameters
----------
labels : list[str]
Label names.
label_type_name : str
Label type.
"""
# checking if label type exists
label_types = self._perfdb.labels.types.get_ids()
if label_type_name not in label_types:
raise ValueError(f"Label type {label_type_name} does not exist in the database")
query = sql.SQL(
"""
DELETE FROM performance.labels
WHERE name IN ({labels}) AND label_type_id = {label_type_id}
""",
).format(
labels=sql.SQL(", ").join(sql.Literal(label) for label in labels),
label_type_id=sql.Literal(label_types[label_type_name]),
)
# deleting data
with self._perfdb.conn.reconnect() as conn:
# deleting
result = conn.execute(query)
logger.debug(f"Deleted {result.rowcount} rows from labels table")
get(label_types=None, output_type='dict')
¶
Gets all label values with detailed information.
The most useful keys/columns returned are:
- id
- label_type_name
- description
Parameters:
-
(label_types¶list[str], default:None) –Types of the labels. If None, retrieves all labels. 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_value is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_value is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
Source code in echo_postgres/label_values.py
@validate_call
def get(
self,
label_types: list[str] | None = None,
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, str | int]] | DataFrame:
"""Gets all label values with detailed information.
The most useful keys/columns returned are:
- id
- label_type_name
- description
Parameters
----------
label_types : list[str], optional
Types of the labels. If None, retrieves all labels. 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_value is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
DataFrame
In case output_value is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
"""
# checking inputs
where = self._check_get_args(label_types=label_types)
if output_type not in ["dict", "DataFrame"]:
raise ValueError(f"output_type must be one of ['dict', 'DataFrame'], not {output_type}")
# building query
query = [sql.SQL("SELECT * FROM performance.v_labels "), where, sql.SQL(" ORDER BY label_name")]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query, post_convert="pyarrow")
df = df.set_index("label_name")
return df.to_dict(orient="index") if output_type == "dict" else df
get_ids(label_types=None)
¶
Gets all label values and their respective ids.
Parameters:
-
(label_types¶list[str], default:None) –Types of the labels. If None, retrieves all labels. By default None.
Returns:
-
dict[str, int]–Dictionary with all label values and their respective ids in the format {label: id, ...}.
Source code in echo_postgres/label_values.py
@validate_call
def get_ids(self, label_types: list[str] | None = None) -> dict[str, int]:
"""Gets all label values and their respective ids.
Parameters
----------
label_types : list[str], optional
Types of the labels. If None, retrieves all labels. By default None.
Returns
-------
dict[str, int]
Dictionary with all label values and their respective ids in the format {label: id, ...}.
"""
# checking inputs
where = self._check_get_args(label_types=label_types)
query = [sql.SQL("SELECT label_name, id FROM performance.v_labels "), where, sql.SQL(" ORDER BY label_name")]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
return df.set_index("label_name").to_dict()["id"]
insert(labels, label_type_name)
¶
Inserts a label value.
Parameters:
Returns:
-
list[int]–List of ids of the inserted labels.
Source code in echo_postgres/label_values.py
@validate_call
def insert(
self,
labels: list[str],
label_type_name: str,
) -> list[int]:
"""Inserts a label value.
Parameters
----------
labels : list[str]
Label names.
label_type_name : str
Label type.
Returns
-------
list[int]
List of ids of the inserted labels.
"""
# checking if label type exists
label_types = self._perfdb.labels.types.get_ids()
if label_type_name not in label_types:
raise ValueError(f"Label type {label_type_name} does not exist in the database")
query = sql.SQL(
"""
INSERT INTO performance.labels (name, label_type_id)
VALUES {values}
ON CONFLICT (name, label_type_id) DO NOTHING
RETURNING id
""",
).format(
values=sql.SQL(", ").join(
sql.SQL("({label}, {label_type_id})").format(
label=sql.Literal(label),
label_type_id=sql.Literal(label_types[label_type_name]),
)
for label in labels
),
)
with self._perfdb.conn.reconnect() as conn:
result = conn.execute(query)
return [row[0] for row in result]