Object Instances¶
ObjectInstances(baze)
¶
Class used for handling object instances.
Parameters:
Source code in echo_baze/object_instances.py
def __init__(self, baze: e_bz.Baze) -> None:
"""Class used for handling Objects.
Parameters
----------
baze : Baze
Top level object carrying all functionality and the connection handler.
"""
super().__init__(baze)
# * subclasses
from .object_instance_attributes import ObjectInstanceAttributes
self.attributes = ObjectInstanceAttributes(baze)
get(output_type='dict', object_names=None, object_ids=None, object_types=None, object_models=None, owners=None, parent_objects=None, get_parent_names=False)
¶
Gets all objects with detailed information.
The most useful keys/columns returned are:
- objectId
- attributes
- objectTypeId
- objectCategoryId
- domainId
- assetId
- owners
- parentId
- parentName (if get_parent_names is True)
- immediateParentId
- immediateParentName (if get_parent_names is True)
Parameters:
-
(output_type¶Literal['dict', 'DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"
-
(object_names¶list[str] | None, default:None) –Wanted object names, by default None. Only one of object_names or object_ids can be used.
-
(object_ids¶list[str] | None, default:None) –Wanted object ids, by default None. Only one of object_names or object_ids can be used.
-
(object_types¶list[str] | None, default:None) –Wanted object types, by default None
-
(object_models¶list[str] | None, default:None) –Wanted object models, by default None
-
(owners¶list[str] | None, default:None) –Wanted owners, by default None
-
(parent_objects¶list[str] | None, default:None) –Wanted parent objects, by default None
-
(get_parent_names¶bool, default:False) –If True it will return the parent names as well, by default False
Returns:
-
dict[str, dict[str, Any]]–In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}
-
DataFrame–In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]
Source code in echo_baze/object_instances.py
@validate_call
def get(
self,
output_type: Literal["dict", "DataFrame"] = "dict",
object_names: list[str] | None = None,
object_ids: list[str] | None = None,
object_types: list[str] | None = None,
object_models: list[str] | None = None,
owners: list[str] | None = None,
parent_objects: list[str] | None = None,
get_parent_names: bool = False,
) -> dict[str, dict[str, Any]] | DataFrame:
"""Gets all objects with detailed information.
The most useful keys/columns returned are:
- objectId
- attributes
- objectTypeId
- objectCategoryId
- domainId
- assetId
- owners
- parentId
- parentName (if get_parent_names is True)
- immediateParentId
- immediateParentName (if get_parent_names is True)
Parameters
----------
output_type : Literal["dict", "DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame"]
By default "dict"
object_names : list[str] | None, optional
Wanted object names, by default None. Only one of object_names or object_ids can be used.
object_ids : list[str] | None, optional
Wanted object ids, by default None. Only one of object_names or object_ids can be used.
object_types : list[str] | None, optional
Wanted object types, by default None
object_models : list[str] | None, optional
Wanted object models, by default None
owners : list[str] | None, optional
Wanted owners, by default None
parent_objects : list[str] | None, optional
Wanted parent objects, by default None
get_parent_names : bool, optional
If True it will return the parent names as well, by default False
Returns
-------
dict[str, dict[str, Any]]
In case output_type == "dict" it will return a dict with the following format: {name: {attribute: value, ...}, ...}
DataFrame
In case output_type == "DataFrame" it will return a DataFrame with the following format: index = name, columns = [attribute, ...]
"""
# checking input
payload, n_filters, obj_type_ids, obj_model_ids = self._check_inputs(
object_ids=object_ids,
object_names=object_names,
object_types=object_types,
object_models=object_models,
owners=owners,
parent_objects=parent_objects,
)
# getting data if object_names is None
if payload is None:
endpoint = "objects/structure"
# getting the data
obj_result = self.baze.conn.get(endpoint)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()["data"]
# getting only the desired objects if object_names is not None (way faster for one object)
else:
endpoint = "objects/list" if "ObjectIds" in payload else "objects"
obj_result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()
if "ObjectIds" in payload:
objs_list = objs_list["objects"]
# adjusting data types
objs_list = self._cast_attributes(objs_list)
# filtering the results
result = self._filter_results(
objs_list,
n_filters,
object_names=object_names,
object_ids=object_ids,
object_types=object_types,
owners=owners,
parent_objects=parent_objects,
object_models=object_models,
)
# getting object type name
obj_type_ids = {v: k for k, v in obj_type_ids.items()}
# getting object model name
obj_model_ids = {v: k for k, v in obj_model_ids.items()}
# adding object type and model names to the attributes
for entry in result:
if "objectTypeId" in entry["attributes"]:
entry["attributes"]["objectTypeName"] = obj_type_ids[entry["attributes"]["objectTypeId"]]
if "domainId" in entry["attributes"]:
entry["attributes"]["domainName"] = obj_model_ids[entry["attributes"]["domainId"]]
# adding parent names if desired
if get_parent_names:
# getting names of the parents
wanted_ids = [x["attributes"]["parentId"] for x in result if "parentId" in x["attributes"]]
wanted_ids += [x["attributes"]["immediateParentId"] for x in result if "immediateParentId" in x["attributes"]]
wanted_ids = list(set(wanted_ids))
# getting the names
parent_objs = self.get_ids(object_ids=wanted_ids)
# adding the names to the result
id_to_name = {v: k for k, v in parent_objs.items()}
for entry in result:
if "parentId" in entry["attributes"]:
entry["attributes"]["parentName"] = id_to_name.get(entry["attributes"]["parentId"], None)
if "immediateParentId" in entry["attributes"]:
entry["attributes"]["immediateParentName"] = id_to_name.get(entry["attributes"]["immediateParentId"], None)
# returning on the desired format
match output_type:
case "dict":
# defining the name as the key for the dict
result_dict = {entry["attributes"]["objectKey"]: entry for entry in result}
# sort the dict by the objectKey
result_dict = dict(sorted(result_dict.items(), key=lambda x: x[0]))
# removing the name from the dict values
for entry in result_dict.values():
del entry["attributes"]["objectKey"]
return result_dict
case "DataFrame":
df = json_normalize(result, max_level=1)
df = df.convert_dtypes(dtype_backend="pyarrow")
df = df.set_index("attributes.objectKey")
df = df.sort_index(ascending=True)
return df
case _:
raise ValueError(f"Invalid output_type: {output_type}")
get_asset_ids(object_names=None, object_ids=None, object_types=None, object_models=None, owners=None, parent_objects=None)
¶
Method that returns a dictionary with the object keys and their asset ids.
Parameters:
-
(object_names¶list[str] | None, default:None) –Wanted object names, by default None. Only one of object_names or object_ids can be used.
-
(object_ids¶list[str] | None, default:None) –Wanted object ids, by default None. Only one of object_names or object_ids can be used.
-
(object_types¶list[str] | None, default:None) –Wanted object types, by default None
-
(object_models¶list[str] | None, default:None) –Wanted object models, by default None
-
(owners¶list[str] | None, default:None) –Wanted owners, by default None
-
(parent_objects¶list[str] | None, default:None) –Wanted parent objects, by default None
Returns:
-
dict[str, int]:–Dictionary with all objects in the format {objectKey: asset_id, ...}
Source code in echo_baze/object_instances.py
@validate_call
def get_asset_ids(
self,
object_names: list[str] | None = None,
object_ids: list[str] | None = None,
object_types: list[str] | None = None,
object_models: list[str] | None = None,
owners: list[str] | None = None,
parent_objects: list[str] | None = None,
) -> dict[str, int]:
"""Method that returns a dictionary with the object keys and their asset ids.
Parameters
----------
object_names : list[str] | None, optional
Wanted object names, by default None. Only one of object_names or object_ids can be used.
object_ids : list[str] | None, optional
Wanted object ids, by default None. Only one of object_names or object_ids can be used.
object_types : list[str] | None, optional
Wanted object types, by default None
object_models : list[str] | None, optional
Wanted object models, by default None
owners : list[str] | None, optional
Wanted owners, by default None
parent_objects : list[str] | None, optional
Wanted parent objects, by default None
Returns
-------
dict[str, int]:
Dictionary with all objects in the format {objectKey: asset_id, ...}
"""
# checking input
payload, n_filters, _, _ = self._check_inputs(
object_names=object_names,
object_ids=object_ids,
object_types=object_types,
object_models=object_models,
owners=owners,
parent_objects=parent_objects,
)
# getting data if object_names is None
if payload is None:
endpoint = "objects/structure"
# getting the data
obj_result = self.baze.conn.get(endpoint)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()["data"]
# getting only the desired objects if object_names is not None (way faster)
else:
endpoint = "objects/list" if "ObjectIds" in payload else "objects"
obj_result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()
if "ObjectIds" in payload:
objs_list = objs_list["objects"]
# adjusting data types
objs_list = self._cast_attributes(objs_list)
# filtering the results
result_list = self._filter_results(
objs_list,
n_filters,
object_names=object_names,
object_ids=object_ids,
object_types=object_types,
owners=owners,
parent_objects=parent_objects,
object_models=object_models,
)
# converting to desired format
result_list = {entry["attributes"]["objectKey"]: entry["attributes"].get("objectHandle", None) for entry in result_list}
return result_list
get_ids(object_names=None, object_ids=None, object_types=None, object_models=None, owners=None, parent_objects=None)
¶
Gets a dictionary with all object keys and ids (string unique id).
Parameters:
-
(object_names¶list[str] | None, default:None) –Wanted object names, by default None. Only one of object_names or object_ids can be used.
-
(object_ids¶list[str] | None, default:None) –Wanted object ids, by default None. Only one of object_names or object_ids can be used.
-
(object_types¶list[str] | None, default:None) –Wanted object types, by default None
-
(object_models¶list[str] | None, default:None) –Wanted object models, by default None
-
(owners¶list[str] | None, default:None) –Wanted owners, by default None
-
(parent_objects¶list[str] | None, default:None) –Wanted parent objects, by default None
Returns:
-
dict[str, UUID]:–Dictionary with all objects in the format {objectKey: UUID, ...}
Source code in echo_baze/object_instances.py
@validate_call
def get_ids(
self,
object_names: list[str] | None = None,
object_ids: list[str] | None = None,
object_types: list[str] | None = None,
object_models: list[str] | None = None,
owners: list[str] | None = None,
parent_objects: list[str] | None = None,
) -> dict[str, UUID]:
"""Gets a dictionary with all object keys and ids (string unique id).
Parameters
----------
object_names : list[str] | None, optional
Wanted object names, by default None. Only one of object_names or object_ids can be used.
object_ids : list[str] | None, optional
Wanted object ids, by default None. Only one of object_names or object_ids can be used.
object_types : list[str] | None, optional
Wanted object types, by default None
object_models : list[str] | None, optional
Wanted object models, by default None
owners : list[str] | None, optional
Wanted owners, by default None
parent_objects : list[str] | None, optional
Wanted parent objects, by default None
Returns
-------
dict[str, UUID]:
Dictionary with all objects in the format {objectKey: UUID, ...}
"""
# checking input
payload, n_filters, _, _ = self._check_inputs(
object_names=object_names,
object_ids=object_ids,
object_types=object_types,
object_models=object_models,
owners=owners,
parent_objects=parent_objects,
)
# getting data if object_names is None
if payload is None:
endpoint = "objects/structure"
# getting the data
obj_result = self.baze.conn.get(endpoint)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()["data"]
else:
endpoint = "objects/list" if "ObjectIds" in payload else "objects"
obj_result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(obj_result)
# converting to dict
objs_list: list[dict[str, Any]] = obj_result.json()
if "ObjectIds" in payload:
objs_list = objs_list["objects"]
# adjusting data types
objs_list = self._cast_attributes(objs_list)
# filtering the results
result_list = self._filter_results(
objs_list,
n_filters,
object_names=object_names,
object_ids=object_ids,
object_types=object_types,
owners=owners,
parent_objects=parent_objects,
object_models=object_models,
)
# converting to desired format
result_list = {entry["attributes"]["objectKey"]: entry["objectId"] for entry in result_list}
return result_list