Attributes¶
Attributes(perfdb)
¶
Class used for handling attribute definitions. Can be accessed via perfdb.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(name)
¶
Deletes an attribute definition.
Parameters:
-
(name¶str) –Name of the attribute definition.
Source code in echo_postgres/attributes.py
@validate_call
def delete(self, name: str) -> None:
"""Deletes an attribute definition.
Parameters
----------
name : str
Name of the attribute definition.
"""
# defining query
query = sql.SQL("DELETE FROM performance.attributes_def WHERE name = {name}").format(name=sql.Literal(name))
# executing query
with self._perfdb.conn.reconnect() as conn:
# deleting
result = conn.execute(query)
logger.debug(f"Deleted {result.rowcount} rows from performance.attributes_def")
get(attribute_names=None, output_type='dict')
¶
Gets all attribute definitions with detailed information.
The most useful keys/columns returned are:
- id
- display_name
- description
- data_type_name
Parameters:
-
(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 is "dict", returns a dictionary in the format {attribute_name: {attribute: value, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = attribute_name, columns = [attribute, ...]
Source code in echo_postgres/attributes.py
@validate_call
def get(
self,
attribute_names: list[str] | None = None,
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | DataFrame:
"""Gets all attribute definitions with detailed information.
The most useful keys/columns returned are:
- id
- display_name
- description
- data_type_name
Parameters
----------
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 is "dict", returns a dictionary in the format {attribute_name: {attribute: value, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = attribute_name, columns = [attribute, ...]
"""
# checking inputs
where = self._check_get_args(attribute_names)
query = [sql.SQL("SELECT * FROM performance.v_attributes_def"), 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")
return df.to_dict(orient="index") if output_type == "dict" else df
get_ids(attribute_names=None)
¶
Gets all attribute definitions and their respective ids.
Parameters:
-
(attribute_names¶list[str] | None, default:None) –List of attribute names to get the ids for. If None, all attribute definitions will be returned, by default None
Returns:
-
dict[str, int]–Dictionary with all attribute definitions and their respective ids in the format {name: id, ...}.
Source code in echo_postgres/attributes.py
@validate_call
def get_ids(self, attribute_names: list[str] | None = None) -> dict[str, int]:
"""Gets all attribute definitions and their respective ids.
Parameters
----------
attribute_names : list[str] | None, optional
List of attribute names to get the ids for. If None, all attribute definitions will be returned, by default None
Returns
-------
dict[str, int]
Dictionary with all attribute definitions and their respective ids in the format {name: id, ...}.
"""
# checking inputs
where = self._check_get_args(attribute_names)
query = [sql.SQL("SELECT name, id FROM performance.v_attributes_def"), 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(name, data_type_name, display_name=None, description=None, on_conflict='raise')
¶
Inserts a new attribute definition.
Parameters:
-
(name¶str) –Name of the attribute definition.
-
(data_type_name¶str) –Name of the data type of the attribute definition.
-
(display_name¶str | None, default:None) –Prettier name used for interfaces with the end user, by default None
-
(description¶str | None, default:None) –Description of the attribute, by default None
-
(on_conflict¶Literal['raise', 'ignore', 'update'], default:'raise') –Determines what will be done in case there is a conflict. Can be one of:
- raise: raises an error
- ignore: ignores the conflict and does nothing
- update: updates the conflicting row with the new data
Source code in echo_postgres/attributes.py
@validate_call
def insert(
self,
name: str,
data_type_name: str,
display_name: str | None = None,
description: str | None = None,
on_conflict: Literal["raise", "ignore", "update"] = "raise",
) -> None:
"""Inserts a new attribute definition.
Parameters
----------
name : str
Name of the attribute definition.
data_type_name : str
Name of the data type of the attribute definition.
display_name : str | None, optional
Prettier name used for interfaces with the end user, by default None
description : str | None, optional
Description of the attribute, by default None
on_conflict : Literal["raise", "ignore", "update"], optional
Determines what will be done in case there is a conflict. Can be one of:
- raise: raises an error
- ignore: ignores the conflict and does nothing
- update: updates the conflicting row with the new data
"""
# checking for allowed data types
allowed_data_types = self._perfdb.datatypes.get_ids()
if data_type_name not in allowed_data_types:
raise ValueError(f"data_type_name must be one of {list(allowed_data_types.keys())}, not {data_type_name}")
# defining query
query = [
sql.SQL(
"INSERT INTO performance.attributes_def (name, data_type_id, display_name, description) "
"VALUES ({name}, {data_type_id}, {display_name}, {description}) ",
).format(
name=sql.Literal(name),
data_type_id=sql.Literal(allowed_data_types[data_type_name]),
display_name=sql.Literal(display_name),
description=sql.Literal(description),
),
]
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 (name) DO UPDATE SET "
"data_type_id = EXCLUDED.data_type_id, "
"display_name = EXCLUDED.display_name, "
"description = EXCLUDED.description",
),
)
# executing query
with self._perfdb.conn.reconnect() as conn:
conn.execute(sql.Composed(query))
logger.debug(f"Attribute definition '{name}' inserted")
update(name, data_type_name=None, display_name=None, description=None)
¶
Updates an attribute definition. Values set to None will not be updated.
Parameters:
-
(name¶str) –Name of the attribute definition.
-
(data_type_name¶str | None, default:None) –Name of the data type of the attribute definition, by default None
-
(display_name¶str | None, default:None) –Prettier name used for interfaces with the end user, by default None
-
(description¶str | None, default:None) –Description of the attribute, by default None
Source code in echo_postgres/attributes.py
@validate_call
def update(
self,
name: str,
data_type_name: str | None = None,
display_name: str | None = None,
description: str | None = None,
) -> None:
"""Updates an attribute definition. Values set to None will not be updated.
Parameters
----------
name : str
Name of the attribute definition.
data_type_name : str | None, optional
Name of the data type of the attribute definition, by default None
display_name : str | None, optional
Prettier name used for interfaces with the end user, by default None
description : str | None, optional
Description of the attribute, by default None
"""
# checking if name exists
ids = self.get_ids()
if name not in ids:
raise ValueError(f"Attribute definition '{name}' does not exist")
# checking if arguments are strings
args = {"data_type_name": data_type_name, "display_name": display_name, "description": description}
# checking if at least one argument is not None
if all(arg is None for arg in args.values()):
raise ValueError("At least one argument must be not None")
# converting data_type_name to id
if data_type_name is not None:
allowed_data_types = self._perfdb.datatypes.get_ids()
if data_type_name not in allowed_data_types:
raise ValueError(f"data_type_name must be one of {list(allowed_data_types.keys())}, not {data_type_name}")
args["data_type_id"] = allowed_data_types[data_type_name]
args.pop("data_type_name")
query = [sql.SQL("UPDATE performance.attributes_def SET ")]
update_query = [
sql.SQL("{arg_name} = {arg_value}").format(
arg_name=sql.Identifier(arg_name),
arg_value=sql.Literal(arg_value),
)
for arg_name, arg_value in args.items()
if arg_value is not None
]
query.append(sql.SQL(", ").join(update_query))
query.append(
sql.SQL(" WHERE name = {name}").format(name=sql.Literal(name)),
)
with self._perfdb.conn.reconnect() as conn:
conn.execute(sql.Composed(query))
logger.debug(f"Attribute definition '{name}' updated")