Hiven Parsers¶
openhivenpy.events.event_parsers.HivenParsers
¶
Event Parsers for Hiven Events that validate and update the cached data
Attributes¶
storage: Optional[ClientCache]
property
readonly
¶
Returns the cached storage
Methods¶
__init__(self, client)
special
¶
Source code in openhivenpy\events\event_parsers.py
def __init__(self, client):
self.client: HivenClient = client
dispatch(self, event, data)
async
¶
Dispatches the parser and returns the args and kwargs. Note that this will only add the event to the buffer and NOT execute it. The asyncio event loop will run it as soon as the
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event |
str |
Event name that should be called |
required |
data |
dict |
Raw WebSocket Data that should be passed |
required |
Returns:
| Type | Description |
|---|---|
Tuple[list, dict] |
The args and kwargs that were created with the parser |
Source code in openhivenpy\events\event_parsers.py
async def dispatch(self, event: str, data: dict) -> Tuple[list, dict]:
"""
Dispatches the parser and returns the args and kwargs. Note that this
will only add the event to the buffer and NOT execute it. The asyncio
event loop will run it as soon as the
:param event: Event name that should be called
:param data: Raw WebSocket Data that should be passed
:return: The args and kwargs that were created with the parser
"""
# getting the method from self
coro = getattr(self, format_event_as_listener(event), None)
if callable(coro):
new_data: dict = deepcopy(data)
return await coro(new_data)
else:
logger.warning(f"[EVENTS] Parser for event {event} was not found!")
on_batch_house_member_update(self, data)
async
¶
LISTENER: on_batch_house_member_update(house: House, members: List[Member])
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_batch_house_member_update(
self, data: dict
) -> Tuple[Tuple, Dict]:
"""
EVENT: BATCH_HOUSE_MEMBER_UPDATE
LISTENER: on_batch_house_member_update(house: House, members: List[Member])
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
house: types.House = self.client.get_house('house_id')
members: List[types.Member] = []
members_data: Dict = data.get('data')
# Updating for every entry and appending the item
for _, mem_data in members_data.items():
self.client.storage.add_or_update_house_member(mem_data)
members.append(
self.client.get_house_member(mem_data['user_id'], house.id)
)
buffer = self._get_from_client_buffer('batch_house_member_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.House, List[types.Member]] = (house, members)
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_down(self, data)
async
¶
LISTENER: house_delete(house_id: str)
house_down(house_id: str)
If property 'unavailable' of the dictionary is false, the listener
house_delete is selected instead of house_down
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_down(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_DOWN
LISTENER:
house_delete(house_id: str)
house_down(house_id: str)
If property 'unavailable' of the dictionary is *false*, the listener
`house_delete` is selected instead of `house_down`
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
if data.get('unavailable') is True:
buffer = self._get_from_client_buffer('house_down')
else:
buffer = self._get_from_client_buffer('house_delete')
self.storage.remove_house(data['house_id'])
# Parameter that will be passed to the assigned listener
args: Tuple = tuple([data['house_id']])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_entities_update(self, data)
async
¶
LISTENER: on_house_entities_update(entities: List[Entity])
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_entities_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_ENTITIES_UPDATE
LISTENER: on_house_entities_update(entities: List[Entity])
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
entities: List[types.Entity] = []
for i in data.get('entities'):
self.client.storage.add_or_update_entity(data)
entities.append(self.client.get_entity(i['id']))
buffer = self._get_from_client_buffer('house_entity_update')
# Parameter that will be passed to the assigned listener
args: Tuple[List[types.Entity]] = tuple([entities])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_join(self, data)
async
¶
LISTENER: on_house_join(house: House)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_join(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_JOIN
LISTENER: on_house_join(house: House)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.storage.add_or_update_house(data)
new_house_data = types.House.format_obj_data(data)
new_house = types.House(new_house_data, self.client)
# Parameter that will be passed to the assigned listener
args: Tuple[types.House] = \
tuple([new_house]) # avoiding warning bc of singular tuple item
kwargs: Dict = {}
buffer = self._get_from_client_buffer('house_join')
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_leave(self, data)
async
¶
LISTENER: on_house_leave(house: House)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_leave(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_LEAVE
LISTENER: on_house_leave(house: House)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.storage.remove_house(data['house_id'])
# Parameter that will be passed to the assigned listener
args: Tuple = tuple([data['house_id']])
kwargs: Dict = {}
buffer = self._get_from_client_buffer('house_leave')
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_member_enter(self, data)
async
¶
LISTENER: on_house_member_online(member: id)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_member_enter(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBER_ENTER
LISTENER: on_house_member_online(member: id)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
user_id: str = data['user_id'] if data.get('user_id') \
else data.get('user', {}).get('id')
self.client.storage.add_or_update_house_member(data)
mem: types.Member = self.client.get_house_member(
user_id, data['house_id']
)
buffer = self._get_from_client_buffer('house_member_online')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Member] = tuple([mem])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_member_exit(self, data)
async
¶
LISTENER: on_house_member_offline(member: id)
House user went offline. Triggers in every house the client, and the user is in the event
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_member_exit(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBER_EXIT
LISTENER: on_house_member_offline(member: id)
House user went offline. Triggers in every house the client,
and the user is in the event
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
mem: types.Member = self.client.get_house_member(
data['id'], data['house_id']
)
buffer = self._get_from_client_buffer('house_member_offline')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Member] = tuple([mem])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_member_join(self, data)
async
¶
LISTENER: on_house_member_join(member: Member)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_member_join(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBER_JOIN
LISTENER: on_house_member_join(member: Member)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.client.storage.add_or_update_house_member(data)
mem: types.Member = self.client.get_house_member(
data['user']['id'], data['house_id']
)
buffer = self._get_from_client_buffer('house_member_join')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Member] = tuple([mem])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_member_leave(self, data)
async
¶
LISTENER: on_house_member_leave(member: Member)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_member_leave(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBER_LEAVE
LISTENER: on_house_member_leave(member: Member)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
mem_id: str = data['user']['id']
house_id: str = data['house_id']
mem: types.Member = self.client.get_house_member(mem_id, house_id)
self.client.storage.remove_house_member(mem_id, house_id)
buffer = self._get_from_client_buffer('house_member_leave')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Member] = tuple([mem])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_member_offline(self, data)
¶
LISTENER: on_house_member_offline(member: id)
Alias for on_house_member_exit
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Coroutine |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
def on_house_member_offline(self, data: dict) -> Coroutine:
"""
EVENT: HOUSE_MEMBER_EXIT
LISTENER: on_house_member_offline(member: id)
Alias for on_house_member_exit
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
# Having to add the self item since it was decorated (special decorator)
return self.on_house_member_exit(self, data)
on_house_member_online(self, data)
¶
LISTENER: on_house_member_online(member: id)
Alias for on_house_member_enter
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Coroutine |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
def on_house_member_online(self, data: dict) -> Coroutine:
"""
EVENT: HOUSE_MEMBER_ENTER
LISTENER: on_house_member_online(member: id)
Alias for on_house_member_enter
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
# Having to add the self item since it was decorated (special decorator)
return self.on_house_member_enter(self, data)
on_house_member_update(self, data)
async
¶
LISTENER: on_house_member_update(member: Member)
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_member_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBER_UPDATE
LISTENER: on_house_member_update(member: Member)
:returns: Args and Kwargs generated by the Parser
"""
user_id: str = data['user_id'] if data.get('user_id') \
else data.get('user', {}).get('id')
self.client.storage.add_or_update_house_member(data)
mem: types.Member = self.client.get_house_member(
user_id, data['house_id']
)
buffer = self._get_from_client_buffer('house_member_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Member] = tuple([mem])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_members_chunk(self, data)
async
¶
LISTENER: on_house_members_chunk(house: House, members: List[Member])
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_members_chunk(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_MEMBERS_CHUNK
LISTENER: on_house_members_chunk(house: House, members: List[Member])
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
user_id = data['user_id'] if data.get('user_id') \
else data.get('user', {}).get('id')
house: types.House = self.client.get_house('house_id')
members: List[types.Member] = []
members_data: Dict = data.get('members')
# Updating for every entry and appending the item
for _, mem_data in members_data.items():
self.client.storage.add_or_update_house_member(mem_data)
members.append(
self.client.get_house_member(user_id, house.id)
)
buffer = self._get_from_client_buffer('house_members_chunk')
# Parameter that will be passed to the assigned listener
args: Tuple[types.House, List[types.Member]] = (house, members)
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_house_update(self, data)
async
¶
LISTENER: on_house_update(old_house: House, new_house: House)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_house_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: HOUSE_UPDATE
LISTENER: on_house_update(old_house: House, new_house: House)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
old_house_data = self.storage['houses'][data['id']]
old_house = types.House(old_house_data, self.client)
new_house_data = self.storage.add_or_update_house(data)
new_house = types.House(new_house_data, self.client)
# Parameter that will be passed to the assigned listener
args: Tuple[types.House, types.House] = (old_house, new_house)
kwargs: Dict = {}
buffer = self._get_from_client_buffer('house_update')
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_message_create(self, data)
async
¶
LISTENER: on_message_create(msg: Message)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_message_create(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: MESSAGE_CREATE
LISTENER: on_message_create(msg: Message)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
msg_data = types.Message.format_obj_data(data)
msg = types.Message(msg_data, self.client)
buffer = self._get_from_client_buffer('message_create')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Message] = tuple([msg])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_message_delete(self, data)
async
¶
LISTENER: on_message_delete(msg_id: str, room_id: str, house_id: str)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_message_delete(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: MESSAGE_DELETE
LISTENER: on_message_delete(msg_id: str, room_id: str, house_id: str)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
buffer = self._get_from_client_buffer('message_delete')
# Parameter that will be passed to the assigned listener
args: Tuple[str, str, str] = (
data.get('message_id'),
data.get('room_id'),
data.get('house_id')
)
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_message_update(self, data)
async
¶
LISTENER: on_message_update(msg: Message)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_message_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: MESSAGE_UPDATE
LISTENER: on_message_update(msg: Message)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
msg_data = types.Message.format_obj_data(data)
msg = types.Message(msg_data, self.client)
buffer = self._get_from_client_buffer('message_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Message] = tuple([msg])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_presence_update(self, data)
async
¶
LISTENER: on_presence_update(user: User)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_presence_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: PRESENCE_UPDATE
LISTENER: on_presence_update(user: User)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.client.storage.add_or_update_user(data)
user: types.User = self.client.get_user(data['id'])
buffer = self._get_from_client_buffer('presence_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.User] = tuple([user])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_relationship_update(self, data)
async
¶
LISTENER: on_relationship_update(relationship: Relationship)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_relationship_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: RELATIONSHIP_UPDATE
LISTENER: on_relationship_update(relationship: Relationship)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.client.storage.add_or_update_relationship(data)
relationship: types.Relationship = self.client.get_relationship(
data['id']
)
buffer = self._get_from_client_buffer('relationship_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.Relationship] = tuple([relationship])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_room_create(self, data)
async
¶
LISTENER: on_room_create(room: Room)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_room_create(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: ROOM_CREATE
LISTENER: on_room_create(room: Room)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.client.storage.add_or_update_room(data)
room: types.TextRoom = self.client.get_room(data['id'])
buffer = self._get_from_client_buffer('room_create')
# Parameter that will be passed to the assigned listener
args: Tuple[types.TextRoom] = tuple([room])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_room_delete(self, data)
async
¶
LISTENER: on_room_delete(room: Room)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_room_delete(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: ROOM_DELETE
LISTENER: on_room_delete(room: Room)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
room: types.TextRoom = self.client.get_room(data['id'])
self.client.storage.remove_room(data['id'])
buffer = self._get_from_client_buffer('room_delete')
# Parameter that will be passed to the assigned listener
args: Tuple[types.TextRoom] = tuple([room])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_room_update(self, data)
async
¶
LISTENER: on_room_update(room: Room)
No data passed at the moment. Gives empty args and kwargs
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_room_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: ROOM_UPDATE
LISTENER: on_room_update(room: Room)
**No data passed at the moment. Gives empty args and kwargs**
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
self.client.storage.add_or_update_room(data)
room: types.TextRoom = self.client.get_room(data['id'])
buffer = self._get_from_client_buffer('room_update')
# Parameter that will be passed to the assigned listener
args: Tuple[types.TextRoom] = tuple([room])
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_typing_start(self, data)
async
¶
LISTENER: on_typing_start(user: User, room: TextRoom, timestamp: datetime.datetime)
No data passed at the moment. Gives empty args and kwargs
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_typing_start(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: TYPING_START
LISTENER: on_typing_start(user: User, room: TextRoom, timestamp: datetime.datetime)
**No data passed at the moment. Gives empty args and kwargs**
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
room_id: str = data['room_id']
if 'recipient_ids' not in data.keys():
room = self.client.get_room(room_id)
else:
if len(list(data['recipient_ids'])) < 2:
room = self.client.get_private_room(room_id)
else:
room = self.client.get_private_group_room(room_id)
user = self.client.get_user(data['author_id'])
# if the timestamp is missing (which never should be the case), the
# time rn will be used.
timestamp = datetime.datetime.fromtimestamp(
safe_convert(int, data.get('timestamp'), time.time()) / 1000
)
buffer = self._get_from_client_buffer('typing_start')
# Parameter that will be passed to the assigned listener
args: Tuple[types.User, BaseRoom, datetime.datetime] = (
user, room, timestamp
)
kwargs: Dict = {}
buffer.add_new_event(data, args, kwargs)
return args, kwargs
on_user_update(self, data)
async
¶
LISTENER: on_user_update(old_user: User, new_user: User)
This parser method modifies the Client cache!
Returns:
| Type | Description |
|---|---|
Tuple[Tuple, Dict] |
Args and Kwargs generated by the Parser |
Source code in openhivenpy\events\event_parsers.py
@log_parser_error()
async def on_user_update(self, data: dict) -> Tuple[Tuple, Dict]:
"""
EVENT: USER_UPDATE
LISTENER: on_user_update(old_user: User, new_user: User)
*This parser method modifies the Client cache!*
:returns: Args and Kwargs generated by the Parser
"""
old_user_data = self.storage['users'][data['id']] # cached data
old_user = types.User(old_user_data, self.client)
user_data = self.storage.add_or_update_user(data)
new_user = types.User(user_data, self.client)
# Parameter that will be passed to the assigned listener
args: Tuple[types.User, types.User] = (old_user, new_user)
kwargs: Dict = {}
buffer: DynamicEventBuffer = self._get_from_client_buffer(
'user_update'
)
buffer.add_new_event(data, args, kwargs)
return args, kwargs