Client Cache¶
openhivenpy.client.cache.ClientCache
¶
Client Cache Class used for storing all data of the Client. Emulates a dictionary and contains additional functions to interact with the Client cache more easily and use functions for better readability.
Methods¶
__init__(self, client, log_websocket, **kwargs)
special
¶
Source code in openhivenpy\client\cache.py
def __init__(self, client: HivenClient, log_websocket: bool, **kwargs):
super(ClientCache, self).__init__(**kwargs)
self.client = client
self.update(
# Updating the passed dict as well to avoid data being overwritten
# that were passed with args or kwargs
utils.update_and_return(
_create_default_dict(log_websocket), **kwargs
)
)
add_or_update_entity(self, item_data)
¶
Adds or Updates a entity to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the Entity class
Source code in openhivenpy\client\cache.py
def add_or_update_entity(self, item_data: dict) -> dict:
"""
Adds or Updates a entity to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the Entity class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['id']
data = types.Entity.format_obj_data(data)
if self['entities'].get(id_) is None:
self['entities'][id_] = data
else:
self['entities'][id_].update(data)
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
add_or_update_house(self, item_data)
¶
Adds or Updates a house to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the House class
Source code in openhivenpy\client\cache.py
def add_or_update_house(self, item_data: dict) -> dict:
"""
Adds or Updates a house to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the House class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['id']
for room in data['rooms']:
room['house_id'] = id_
room = types.TextRoom.format_obj_data(room)
self.add_or_update_room(room)
for member in data['members']:
member['house_id'] = id_
user = types.User.format_obj_data(member['user'])
self.add_or_update_user(user)
for entity in data['entities']:
entity['house_id'] = id_
entity = types.Entity.format_obj_data(entity)
self.add_or_update_entity(entity)
data = types.House.format_obj_data(data)
data['client_member'] = data['members'][self['client_user']['id']]
if self['houses'].get(id_) is None:
self['houses'][id_] = data
else:
self['houses'][id_].update(data)
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
add_or_update_private_room(self, item_data)
¶
Adds or Updates a private room to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the Private_*Room class
Source code in openhivenpy\client\cache.py
def add_or_update_private_room(self, item_data: dict) -> dict:
"""
Adds or Updates a private room to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the Private_*Room class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['id']
if int(data['type']) == 1:
types.PrivateRoom.format_obj_data(data)
if self['rooms']['private']['single'].get(id_) is None:
self['rooms']['private']['single'][id_] = data
else:
self['rooms']['private']['single'][id_].update(data)
elif int(data['type']) == 2:
types.PrivateGroupRoom.format_obj_data(data)
if self['rooms']['private']['group'].get(id_) is None:
self['rooms']['private']['group'][id_] = data
else:
self['rooms']['private']['group'][id_].update(data)
else:
raise ValueError("Data does not contain correct type-id")
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
add_or_update_relationship(self, item_data)
¶
Adds or Updates a client relationship to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the Relationship class
Source code in openhivenpy\client\cache.py
def add_or_update_relationship(self, item_data: dict) -> dict:
"""
Adds or Updates a client relationship to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the Relationship class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['user_id']
if data.get('user'):
self.add_or_update_user(data['user'])
data = types.Relationship.format_obj_data(data)
if self['relationships'].get(id_) is None:
self['relationships'][id_] = data
else:
self['relationships'][id_].update(data)
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
add_or_update_room(self, item_data)
¶
Adds or Updates a room to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the Room class
Source code in openhivenpy\client\cache.py
def add_or_update_room(self, item_data: dict) -> dict:
"""
Adds or Updates a room to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the Room class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['id']
data = types.TextRoom.format_obj_data(data)
if self['rooms']['house'].get(id_) is None:
self['rooms']['house'][id_] = data
else:
self['rooms']['house'][id_].update(data)
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
add_or_update_user(self, item_data)
¶
Adds or Updates a user to the cache and updates the storage appropriately
:return: The validated data using format_obj_data
of the User class
Source code in openhivenpy\client\cache.py
def add_or_update_user(self, item_data: dict) -> dict:
"""
Adds or Updates a user to the cache and updates the storage appropriately
:return: The validated data using `format_obj_data` of the User class
"""
self.check_if_initialised()
try:
data = deepcopy(item_data)
id_ = data['id']
data = types.User.format_obj_data(data)
if id_ == self['client_user'].get('id'):
self.update_client_user(data)
if self['users'].get(id_) is None:
self['users'][id_] = data
else:
self['users'][id_].update(data)
return data
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new house to the Client cache:",
exc_info=sys.exc_info()
)
raise InitializationError("Failed to update the cache due to an exception occurring") from e
check_if_initialised(self)
¶
Checks whether the client has initialised
Source code in openhivenpy\client\cache.py
def check_if_initialised(self) -> bool:
""" Checks whether the client has initialised """
if self.get('client_user'):
return True
else:
raise ValueError("Data Updates require a initialised Hiven Client!")
closing_cleanup(self)
¶
Cleans all remaining data after the client exited.
Not supposed to be called outside of the intended HivenClient.close() method!
Source code in openhivenpy\client\cache.py
def closing_cleanup(self) -> None:
"""
Cleans all remaining data after the client exited.
Not supposed to be called outside of the intended HivenClient.close()
method!
"""
self.update(_create_default_dict(self['log_websocket']))
self['client_user'] = {}
self.init_client_user_obj()
init_client_user_obj(self)
¶
Initialises the client user based on the cached data
Source code in openhivenpy\client\cache.py
def init_client_user_obj(self) -> types.User:
""" Initialises the client user based on the cached data """
return types.User(self['client_user'], self.client)
update_client_user(self, item_data)
¶
Updating the Client Cache Data from the passed data dict
:return: The validated data using format_obj_data
of the User class
Source code in openhivenpy\client\cache.py
def update_client_user(self, item_data: dict) -> dict:
"""
Updating the Client Cache Data from the passed data dict
:return: The validated data using `format_obj_data` of the User class
"""
data = deepcopy(item_data)
client_user = types.User.format_obj_data(data)
self['client_user'].update(client_user)
if self['users'].get(data['id']) is not None:
self['users'][data['id']].update(client_user)
else:
self['users'][data['id']] = client_user
return client_user
update_primary_data(self, item_data)
¶
Updates in the cache the following data: - List of all House Memberships - List of all House Ids - The Client settings of the user - The read state of messages - All open Private Rooms - All Relationships of the user
Source code in openhivenpy\client\cache.py
def update_primary_data(self, item_data: dict) -> None:
"""
Updates in the cache the following data:
- List of all House Memberships
- List of all House Ids
- The Client settings of the user
- The read state of messages
- All open Private Rooms
- All Relationships of the user
"""
data = deepcopy(item_data)
self['house_memberships'] = data.get('house_memberships', {})
self['house_ids'] = data.get('house_ids', [])
self['settings'] = data.get('settings', {})
self['read_state'] = data.get('read_state', {})
self.update_client_user(data.get('user'))
for r in data.get('private_rooms', []):
self.add_or_update_private_room(r)
for key, data in data.get('relationships', []).items():
self.add_or_update_relationship(data)