Component Models - Attributes¶
ComponentModelAttributes(perfdb)
¶
Class used for handling component models attributes. Can be accessed via perfdb.components.models.attributes.
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(component_model_name, attribute_name)
¶
Deletes an attribute value.
Parameters:
-
(component_model_name¶str) –Name of the component model to delete the attribute value from.
-
(attribute_name¶str) –Name of the attribute to delete the value from.
Source code in echo_postgres/component_model_attributes.py
@validate_call
def delete(
self,
component_model_name: str,
attribute_name: str,
) -> None:
"""Deletes an attribute value.
Parameters
----------
component_model_name : str
Name of the component model 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.component_model_attributes "
"WHERE component_model_id = (SELECT id FROM performance.component_models WHERE name = {component_model_name}) "
"AND attribute_id = (SELECT id FROM performance.attributes_def WHERE name = {attribute_name}) ",
).format(
component_model_name=sql.Literal(component_model_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.component_model_attributes")
get(component_models=None, attribute_names=None, filter_type='and', output_type='dict', values_only=False)
¶
Method to get the attributes of the given component models.
The most useful keys/columns returned are:
- component_model_id
- component_model_name
- attribute_id
- attribute_name
- attribute_display_name
- attribute_value
- data_type_name
Parameters:
-
(component_models¶list[str] | None, default:None) –List of component models 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
-
(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 {component_model_name: {attribute_name: {attribute: value, ...}, ...}, ...} If values_only is set to True, returns a dictionary in the format {component_model_name: {attribute_name: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[component_model_name, attribute_name], columns = [attribute, ...] If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[component_model_name, attribute_name], columns = ["attribute_value"]
Source code in echo_postgres/component_model_attributes.py
@validate_call
def get(
self,
component_models: list[str] | None = None,
attribute_names: list[str] | 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 component models.
The most useful keys/columns returned are:
- component_model_id
- component_model_name
- attribute_id
- attribute_name
- attribute_display_name
- attribute_value
- data_type_name
Parameters
----------
component_models : list[str] | None, optional
List of component models 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
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 {component_model_name: {attribute_name: {attribute: value, ...}, ...}, ...}
If values_only is set to True, returns a dictionary in the format {component_model_name: {attribute_name: value, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[component_model_name, attribute_name], columns = [attribute, ...]
If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[component_model_name, attribute_name], columns = ["attribute_value"]
"""
# checking if all component models are valid
if component_models:
existing_models = self._perfdb.components.models.get_ids(component_models=component_models)
if missing_models := set(component_models) - set(existing_models.keys()):
raise ValueError(f"The following component models do not exist: {missing_models}")
# building the query
query = [
sql.SQL(
"SELECT {values} FROM performance.v_component_model_attributes",
).format(
values=sql.SQL(
"component_model_name, attribute_name, attribute_value, data_type_name",
)
if values_only
else sql.SQL("*"),
),
]
where = []
if component_models:
where.append(
sql.SQL(" component_model_name IN ({component_models}) ").format(
component_models=sql.SQL(",").join(sql.Literal(om) for om in component_models),
),
)
if attribute_names:
where.append(
sql.SQL(" attribute_name IN ({attribute_names}) ").format(
attribute_names=sql.SQL(",").join(sql.Literal(an) for an in attribute_names),
),
)
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 component_model_name, attribute_name"))
query = sql.Composed(query)
# executing the query
with self._perfdb.conn.reconnect() as conn:
# setting attribute_value as component 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=["component_model_name"])
df = df.set_index(["component_model_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(component_model_name, attribute_name, attribute_value, on_conflict='raise')
¶
Inserts a new attribute value.
Parameters:
-
(component_model_name¶str) –Name of the component model 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/component_model_attributes.py
@validate_call
def insert(
self,
component_model_name: str,
attribute_name: str,
attribute_value: Any,
on_conflict: Literal["raise", "ignore", "update"] = "raise",
) -> None:
"""Inserts a new attribute value.
Parameters
----------
component_model_name : str
Name of the component model 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 component model exists
existing_models = self._perfdb.components.models.get_ids(component_models=[component_model_name])
if component_model_name not in existing_models:
raise ValueError(f"The component model {component_model_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.component_model_attributes (component_model_id, attribute_id, value) "
"VALUES ({component_model_id}, {attribute_id}, {attribute_value}) ",
).format(
component_model_id=sql.Literal(existing_models[component_model_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 (component_model_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 component model '{component_model_name}'")
update(component_model_name, attribute_name, attribute_value)
¶
Updates an attribute value.
Parameters:
-
(component_model_name¶str) –Name of the component model 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/component_model_attributes.py
@validate_call
def update(
self,
component_model_name: str,
attribute_name: str,
attribute_value: Any,
) -> None:
"""Updates an attribute value.
Parameters
----------
component_model_name : str
Name of the component model 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 component model exists
existing_models = self._perfdb.components.models.get_ids(component_models=[component_model_name])
if component_model_name not in existing_models:
raise ValueError(f"The component model {component_model_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.component_model_attributes "
"SET value = {attribute_value} "
"WHERE component_model_id = {component_model_id} "
"AND attribute_id = {attribute_id} ",
).format(
component_model_id=sql.Literal(existing_models[component_model_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 component model '{component_model_name}'")