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, **kwargs)
special
¶
Source code in openhivenpy\client\cache.py
def __init__(self, client: HivenClient, **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_cache(), **kwargs
)
)
add_or_update_entity(self, item_data)
¶
Adds or updates a entity to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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 entity to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_house(self, item_data)
¶
Adds or updates a house to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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_
self.add_or_update_room(room)
for member in data['members']:
member['house_id'] = id_
# Adding the users already
user = types.User.format_obj_data(member['user'])
self.add_or_update_user(user)
for entity in data['entities']:
entity['house_id'] = id_
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:
# Checking whether id does not already exist
# On init house_ids is already populated using the event
# INIT_STATE, though HOUSE_JOIN triggers this function, meaning
# we have to avoid creating duplicates on initialisation
if id_ not in self['house_ids']:
self['house_ids'].append(id_)
self['houses'][id_] = data
else:
self['houses'][id_].update(data)
# After the House was created altering the cached data
for member in data['members'].values():
self.add_or_update_house_member(member)
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 InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_house_member(self, item_data)
¶
Adds or updates a member inside a House storage
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
Source code in openhivenpy\client\cache.py
def add_or_update_house_member(self, item_data: dict) -> dict:
"""
Adds or updates a member inside a House storage
:return: The validated data using `format_obj_data` of the Member class
"""
self.check_if_initialised()
try:
# If not dict -> User property was replaced -
# house.format_obj_data() replaced it with the ids of the
# corresponding users
if type(item_data['user']) is not dict:
item_data['user'] = self.client.find_user(item_data['user'])
member = types.Member.format_obj_data(item_data)
mem_id = item_data['user_id'] if item_data.get('user_id') \
else item_data.get('user', {}).get('id')
house_id = item_data['house_id']
if mem_id in self['houses'][house_id]['members'].keys():
self['houses'][house_id]['members'][mem_id].update(member)
else:
self['houses'][house_id]['members'][mem_id] = member
user = types.User.format_obj_data(item_data['user'])
self.add_or_update_user(user)
return member
except Exception as e:
utils.log_traceback(
brief=f"Failed to add a new member to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_private_room(self, item_data)
¶
Adds or updates a private room to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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 private room to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_relationship(self, item_data)
¶
Adds or updates a client relationship to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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)
if 'user_id' in data.keys():
id_ = data['user_id']
elif 'id' in data.keys():
id_ = data['id']
elif data.get('user'):
id_ = data['user']['id']
else:
raise InvalidPassedDataError(
"The data does not contain any correct id item"
" or user field",
data=data
) # how?
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 relationship to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_room(self, item_data)
¶
Adds or updates a room to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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 room to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
add_or_update_user(self, item_data)
¶
Adds or updates a user to the cache and updates the storage appropriately
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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 user to the Client cache:",
exc_info=sys.exc_info()
)
raise InvalidPassedDataError(
"Failed to update the cache due to faulty data being passed",
data=item_data
) from e
check_if_initialised(self)
¶
Checks whether the client has initialised
Exceptions:
| Type | Description |
|---|---|
ValueError |
If the client_user is not initialised |
Source code in openhivenpy\client\cache.py
def check_if_initialised(self) -> bool:
"""
Checks whether the client has initialised
:raises ValueError: If the client_user is not initialised
"""
if self.get('client_user'):
return True
else:
raise ValueError("Updates require an 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_cache())
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)
remove_entity(self, _id)
¶
Removes an entity from the cache
Source code in openhivenpy\client\cache.py
def remove_entity(self, _id: str) -> None:
""" Removes an entity from the cache """
self.check_if_initialised()
del self['entities'][_id]
remove_house(self, _id)
¶
Removes a house from the cache
Source code in openhivenpy\client\cache.py
def remove_house(self, _id: str) -> None:
""" Removes a house from the cache """
self.check_if_initialised()
for room in self['houses'][_id]['rooms']:
room: str
self.remove_room(room)
for entity in self['houses'][_id]['entities']:
entity: str
self.remove_entity(entity)
del self['houses'][_id]
self['house_ids']: list
self['house_ids'].remove(_id)
remove_house_member(self, member_id, house_id)
¶
Removes a house from the cache
Source code in openhivenpy\client\cache.py
def remove_house_member(self, member_id: str, house_id: str) -> None:
""" Removes a house from the cache """
self.check_if_initialised()
del self['houses'][house_id]['members'][member_id]
remove_private_room(self, _id)
¶
Removes a private-room from the cache
Source code in openhivenpy\client\cache.py
def remove_private_room(self, _id: str) -> None:
""" Removes a private-room from the cache """
self.check_if_initialised()
for i in self['rooms']['private']['group']:
if i.get('id') == _id:
del self['rooms']['private']['group'][_id]
return
for i in self['rooms']['private']['single']:
if i.get('id') == _id:
del self['rooms']['private']['single'][_id]
remove_relationship(self, _id)
¶
Removes a relationship from the cache
Source code in openhivenpy\client\cache.py
def remove_relationship(self, _id: str) -> None:
""" Removes a relationship from the cache """
self.check_if_initialised()
del self['relationships'][_id]
remove_room(self, _id)
¶
Removes a room from the cache
Source code in openhivenpy\client\cache.py
def remove_room(self, _id: str) -> None:
""" Removes a room from the cache """
self.check_if_initialised()
del self['rooms']['house'][_id]
remove_user(self, _id)
¶
Removes a user from the cache
Source code in openhivenpy\client\cache.py
def remove_user(self, _id: str) -> None:
""" Removes a user from the cache """
self.check_if_initialised()
del self['users'][_id]
update_client_user(self, item_data)
¶
Updating the Client Cache Data from the passed data dict
Returns:
| Type | Description |
|---|---|
dict |
The validated data using |
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_ids'] = data.get('house_ids', [])
self['settings'] = data.get('settings', {})
self['init_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)