Resource Assessment Instances¶
ResourceAssessmentInstances(perfdb)
¶
Class used for handling resource assessment instances. Can be accessed via perfdb.resourceassessments.instances.
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
get(resource_assessment_names=None, resource_assessment_types=None, companies=None, filter_type='and', output_type='dict')
¶
Gets all resource assessment instances definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- resource_assessment_type_name
- company_name
Parameters:
-
(resource_assessment_names¶list[str] | None, default:None) –List of resource assessment instances to filter the query. If None, no filter will be applied. By default None
-
(resource_assessment_types¶list[str] | None, default:None) –List of resource assessment types to filter the query. If None, no filter will be applied. By default None
-
(companies¶list[str] | None, default:None) –List of companies to filter the query. If None, no filter will be applied. By default None
-
(filter_type¶Literal['and', 'or'], default:'and') –Type of filter to apply. 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"
Returns:
-
dict[str, dict[str, Any]]–In case output_instance is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_instance is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
Source code in echo_postgres/resourceassessment_instances.py
@validate_call
def get(
self,
resource_assessment_names: list[str] | None = None,
resource_assessment_types: list[str] | None = None,
companies: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | DataFrame:
"""Gets all resource assessment instances definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- resource_assessment_type_name
- company_name
Parameters
----------
resource_assessment_names : list[str] | None
List of resource assessment instances to filter the query. If None, no filter will be applied. By default None
resource_assessment_types : list[str] | None
List of resource assessment types to filter the query. If None, no filter will be applied. By default None
companies : list[str] | None
List of companies to filter the query. If None, no filter will be applied. By default None
filter_type : Literal["and", "or"], optional
Type of filter to apply. 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"
Returns
-------
dict[str, dict[str, Any]]
In case output_instance is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
DataFrame
In case output_instance is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
"""
if output_type not in ["dict", "DataFrame"]:
raise ValueError(f"output_type must be one of ['dict', 'DataFrame'], not {output_type}")
# validating arguments and building WHERE clause
where = self._check_get_args(
resource_assessment_names=resource_assessment_names,
resource_assessment_types=resource_assessment_types,
companies=companies,
filter_type=filter_type,
)
query = [sql.SQL("SELECT * FROM performance.v_resource_assessments ")]
if where:
query.append(where)
query.append(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(resource_assessment_names=None, resource_assessment_types=None, companies=None, filter_type='and')
¶
Gets all resource assessment instances and their respective ids.
Parameters:
-
(resource_assessment_names¶list[str] | None, default:None) –List of resource assessment instances to filter the query. If None, no filter will be applied. By default None
-
(resource_assessment_types¶list[str] | None, default:None) –List of resource assessment types to filter the query. If None, no filter will be applied. By default None
-
(companies¶list[str] | None, default:None) –List of companies to filter the query. If None, no filter will be applied. By default None
-
(filter_type¶Literal['and', 'or'], default:'and') –Type of filter to apply. Can be one of ["and", "or"]. By default "and"
Returns:
-
dict[str, int]–Dictionary with all resource assessment instances and their respective ids in the format {resource_assessment_instance: id, ...}.
Source code in echo_postgres/resourceassessment_instances.py
@validate_call
def get_ids(
self,
resource_assessment_names: list[str] | None = None,
resource_assessment_types: list[str] | None = None,
companies: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> dict[str, int]:
"""Gets all resource assessment instances and their respective ids.
Parameters
----------
resource_assessment_names : list[str] | None
List of resource assessment instances to filter the query. If None, no filter will be applied. By default None
resource_assessment_types : list[str] | None
List of resource assessment types to filter the query. If None, no filter will be applied. By default None
companies : list[str] | None
List of companies to filter the query. If None, no filter will be applied. By default None
filter_type : Literal["and", "or"], optional
Type of filter to apply. Can be one of ["and", "or"]. By default "and"
Returns
-------
dict[str, int]
Dictionary with all resource assessment instances and their respective ids in the format {resource_assessment_instance: id, ...}.
"""
# validating arguments and building WHERE clause
where = self._check_get_args(
resource_assessment_names=resource_assessment_names,
resource_assessment_types=resource_assessment_types,
companies=companies,
filter_type=filter_type,
)
query = [sql.SQL("SELECT name, id FROM performance.v_resource_assessments ")]
if where:
query.append(where)
query.append(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"]