WebSocket Clients¶
openhivenpy.client.hivenclient.HivenClient
¶
Main Class for connecting to Hiven and interacting with the API.
bot: Optional[bool]
property
readonly
¶
client_type: Optional[str]
property
readonly
¶
client_user: Optional[openhivenpy.types.user.User]
property
readonly
¶
connection: Optional[openhivenpy.gateway.Connection]
property
readonly
¶
connection_status: Optional[int]
property
readonly
¶
header: Optional[str]
property
readonly
¶
http: Optional[openhivenpy.gateway.http.HTTP]
property
readonly
¶
icon: Optional[str]
property
readonly
¶
id: Optional[str]
property
readonly
¶
initialised: Optional[bool]
property
readonly
¶
joined_at: Optional[datetime.datetime]
property
readonly
¶
location: Optional[str]
property
readonly
¶
log_websocket: Optional[str]
property
readonly
¶
loop: Optional[asyncio.events.AbstractEventLoop]
property
readonly
¶
message_broker: Optional[openhivenpy.gateway.messagebroker.MessageBroker]
property
readonly
¶
name: Optional[str]
property
readonly
¶
open: Optional[bool]
property
readonly
¶
presence: Optional[str]
property
readonly
¶
queue_events: Optional[bool]
property
readonly
¶
startup_time: Optional[int]
property
readonly
¶
storage: Optional[openhivenpy.client.cache.ClientCache]
property
readonly
¶
token: Optional[str]
property
readonly
¶
username: Optional[str]
property
readonly
¶
website: Optional[str]
property
readonly
¶
Methods¶
__init__(self, *, token=None, loop=None, log_websocket=False, queue_events=False, host=None, api_version=None, heartbeat=None, close_timeout=None)
special
¶
:param token: Token that can be passed pre-runtime. If not set, the token will need to be passed at run-time. If a token is passed using local available environment variables and no other token is passed that one will be used. :param loop: Loop that will be used to run the Client. If a new one is passed on run() that one will be used instead :param log_websocket: If set to True will additionally log websocket messages and their content :param host: The host API endpoint of Hiven. Defaults to api.hiven.io :param api_version: The API version that should be used. Defaults to v1 :param queue_events: If set to True the received events over the websocket will be queued and event_listeners will called one after another. If set to False all events are directly assigned to the asyncio event_loop and executed parallel :param heartbeat: Intervals in which the bot will send heartbeats to the Websocket. Defaults to the pre-set environment variable heartbeat (default at 30000) :param close_timeout: Seconds after the websocket will timeout after the end handshake didn't complete successfully. Defaults to the pre-set environment variable close_timeout (default at 40)
Source code in openhivenpy\client\hivenclient.py
def __init__(
self,
*,
token: str = None,
loop: AbstractEventLoop = None,
log_websocket: bool = False,
queue_events: bool = False,
host: Optional[str] = None,
api_version: Optional[str] = None,
heartbeat: Optional[int] = None,
close_timeout: Optional[int] = None
):
"""
:param token: Token that can be passed pre-runtime. If not set, the
token will need to be passed at run-time. If a token is passed using
local available environment variables and no other token is passed
that one will be used.
:param loop: Loop that will be used to run the Client. If a new one is
passed on run() that one will be used instead
:param log_websocket: If set to True will additionally log websocket
messages and their content
:param host: The host API endpoint of Hiven. Defaults to api.hiven.io
:param api_version: The API version that should be used. Defaults to v1
:param queue_events: If set to True the received events over the
websocket will be queued and event_listeners will called one after
another. If set to False all events are directly assigned to the
asyncio event_loop and executed parallel
:param heartbeat: Intervals in which the bot will send heartbeats to
the Websocket. Defaults to the pre-set environment variable heartbeat
(default at 30000)
:param close_timeout: Seconds after the websocket will timeout after
the end handshake didn't complete successfully. Defaults to the pre-set
environment variable close_timeout (default at 40)
"""
self._token = token
self._loop = loop
self._log_websocket = log_websocket
self._queue_events = queue_events
self._client_user = None
self._storage = ClientCache(client=self, log_websocket=log_websocket,
token=self._token)
self._connection = Connection(
self, api_version=api_version, host=host, heartbeat=heartbeat,
close_timeout=close_timeout
)
# Inheriting the HivenEventHandler class that will call and trigger
# the parsers for events
super().__init__(client=self, parsers=HivenParsers(self))
__repr__(self)
special
¶
Source code in openhivenpy\client\hivenclient.py
def __repr__(self) -> str:
info = [
('type', self.client_type),
('open', getattr(self, 'open', False)),
('bot', getattr(self, 'bot', 'na')),
('name', getattr(self.client_user, 'name', 'na')),
('id', getattr(self.client_user, 'id', 'na'))
]
return '<{} {}>'.format(self.__class__.__name__, ' '.join('%s=%s' % t for t in info))
__str__(self)
special
¶
Source code in openhivenpy\client\hivenclient.py
def __str__(self) -> str:
return getattr(self, "name")
close(self, force=False, remove_listeners=True)
async
¶
Closes the Connection to Hiven and stops the running WebSocket and the Event Processing Loop
:param force: If set to True the running event-listener workers will be forced closed, which may lead to running code of event-listeners being stopped while performing actions. If False the stopping will wait for all running event_listeners to finish :param remove_listeners: If sett to True, it will remove all listeners including the ones created using @client.event(), add_multi_listener() and add_single_listener()
Source code in openhivenpy\client\hivenclient.py
async def close(
self, force: bool = False, remove_listeners: bool = True
) -> None:
"""
Closes the Connection to Hiven and stops the running WebSocket and
the Event Processing Loop
:param force: If set to True the running event-listener workers will be
forced closed, which may lead to running code of event-listeners being
stopped while performing actions. If False the stopping will wait
for all running event_listeners to finish
:param remove_listeners: If sett to True, it will remove all listeners
including the ones created using @client.event(), add_multi_listener()
and add_single_listener()
"""
await self.connection.close(force, remove_listeners)
self.storage.closing_cleanup()
logger.debug(f"[HIVENCLIENT] Client {repr(self)} was closed")
connect(self, token=None, *, restart=False)
async
¶
Establishes a connection to Hiven and does not return until finished
:param token: Token that should be used to connect to Hiven. If none is passed it will try to fetch the token in the environment variables using os.getenv('HIVEN_TOKEN'). Will overwrite the pre-runtime passed token if one was passed :param restart: If set to True the Client will restart if an error is encountered!
Source code in openhivenpy\client\hivenclient.py
async def connect(
self,
token: str = None,
*,
restart: bool = False
) -> None:
"""Establishes a connection to Hiven and does not return until finished
:param token: Token that should be used to connect to Hiven. If none is
passed it will try to fetch the token in the environment variables
using os.getenv('HIVEN_TOKEN'). Will overwrite the pre-runtime passed
token if one was passed
:param restart: If set to True the Client will restart if an error is
encountered!
"""
try:
if token is None:
token = os.getenv('HIVEN_TOKEN')
if self._token is None and token is not None:
self._token = token
self.storage['token'] = self._token
user_token_len: int = utils.safe_convert(
int, os.getenv("USER_TOKEN_LEN")
)
bot_token_len: int = utils.safe_convert(
int, os.getenv("BOT_TOKEN_LEN")
)
if self._token is None or self._token == "":
logger.critical(f"[HIVENCLIENT] Empty Token was passed!")
raise InvalidTokenError("Empty Token was passed!")
elif len(self._token) not in (user_token_len, bot_token_len):
logger.critical(f"[HIVENCLIENT] Invalid Token was passed")
raise InvalidTokenError("Invalid Token was passed")
await self.connection.connect(restart=restart)
except KeyboardInterrupt:
pass
except (InvalidTokenError, WebSocketFailedError):
raise
except SessionCreateError:
raise
except Exception as e:
utils.log_traceback(
level='critical',
brief=f"Failed to keep alive connection to Hiven:",
exc_info=sys.exc_info()
)
raise HivenConnectionError(
f"Failed to keep alive connection to Hiven"
) from e
edit(self, **kwargs)
async
¶
Edits the Clients data on Hiven
Available options: header, icon, bio, location, website, username
:return: True if the request was successful else False
Source code in openhivenpy\client\hivenclient.py
async def edit(self, **kwargs) -> bool:
"""
Edits the Clients data on Hiven
---
Available options: header, icon, bio, location, website, username
:return: True if the request was successful else False
"""
try:
for key in kwargs.keys():
if key in ['header', 'icon', 'bio', 'location', 'website',
'username']:
await self.http.patch(
endpoint="/users/@me", json={key: kwargs.get(key)}
)
return True
else:
raise NameError("The passed value does not exist in the Client!")
except Exception as e:
keys = "".join(str(key + " ") for key in kwargs.keys())
utils.log_traceback(
brief=f"Failed change the values {keys}:",
exc_info=sys.exc_info()
)
raise
find_entity(self, entity_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param entity_id: id of the Entity :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_entity(self, entity_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param entity_id: id of the Entity
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['entities'].get(entity_id)
if raw_data:
return dict(raw_data)
else:
return None
find_house(self, house_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param house_id: id of the House :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_house(self, house_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param house_id: id of the House
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['houses'].get(house_id)
if raw_data:
return dict(raw_data)
else:
return None
find_private_group_room(self, room_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param room_id: id of the PrivateGroupRoom :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_private_group_room(self, room_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param room_id: id of the PrivateGroupRoom
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['rooms']['private']['group'].get(room_id)
if raw_data:
return dict(raw_data)
else:
return None
find_private_room(self, room_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param room_id: id of the PrivateRoom :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_private_room(self, room_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param room_id: id of the PrivateRoom
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['rooms']['private']['single'].get(room_id)
if raw_data:
return dict(raw_data)
else:
return None
find_relationship(self, user_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param user_id: user-id of the Relationship :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_relationship(self, user_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param user_id: user-id of the Relationship
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['relationships'].get(user_id)
if raw_data:
return dict(raw_data)
else:
return None
find_room(self, room_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param room_id: id of the Room :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_room(self, room_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param room_id: id of the Room
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['rooms']['house'].get(room_id)
if raw_data:
return dict(raw_data)
else:
return None
find_user(self, user_id)
¶
Fetches a dictionary from the cache based on the passed id
The returned dict is only a copy from the cache
:param user_id: id of the User :return: The cached dict if it exists in the cache else None
Source code in openhivenpy\client\hivenclient.py
def find_user(self, user_id: str) -> Optional[dict]:
"""
Fetches a dictionary from the cache based on the passed id
---
The returned dict is only a copy from the cache
:param user_id: id of the User
:return: The cached dict if it exists in the cache else None
"""
raw_data = self.storage['users'].get(user_id)
if raw_data:
return dict(raw_data)
else:
return None
get_entity(self, entity_id)
¶
Fetches a Entity instance from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param entity_id: id of the Entity :return: The Entity instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_entity(self, entity_id: str) -> Optional[types.Entity]:
"""
Fetches a Entity instance from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if
changes are made while the instance exists the data will not be
updated!
:param entity_id: id of the Entity
:return: The Entity instance if it was found else None
"""
raw_data = self.find_entity(entity_id)
if raw_data:
return types.Entity(raw_data, self)
else:
return None
get_house(self, house_id)
¶
Fetches a House from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param house_id: id of the House :return: The house instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_house(self, house_id: str) -> Optional[types.House]:
"""
Fetches a House from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if
changes are made while
the instance exists the data will not be updated!
:param house_id: id of the House
:return: The house instance if it was found else None
"""
raw_data = self.find_house(house_id)
if raw_data:
return types.House(raw_data, self)
else:
return None
get_private_group_room(self, room_id)
¶
Fetches a multi PrivateGroupRoom from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param room_id: id of the PrivateGroupRoom :return: The PrivateGroupRoom instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_private_group_room(self, room_id: str) -> Optional[types.PrivateGroupRoom]:
"""
Fetches a multi PrivateGroupRoom from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if changes are made while
the instance exists the data will not be updated!
:param room_id: id of the PrivateGroupRoom
:return: The PrivateGroupRoom instance if it was found else None
"""
raw_data = self.find_private_group_room(room_id)
if raw_data:
return types.PrivateGroupRoom(raw_data, self)
else:
return None
get_private_room(self, room_id)
¶
Fetches a single PrivateRoom from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param room_id: id of the PrivateRoom :return: The PrivateRoom instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_private_room(self, room_id: str) -> Optional[types.PrivateRoom]:
"""
Fetches a single PrivateRoom from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if changes are made while
the instance exists the data will not be updated!
:param room_id: id of the PrivateRoom
:return: The PrivateRoom instance if it was found else None
"""
raw_data = self.find_private_room(room_id)
if raw_data:
return types.PrivateRoom(raw_data, self)
else:
return None
get_relationship(self, user_id)
¶
Fetches a Relationship instance from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param user_id: user-id of the Relationship :return: The Relationship instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_relationship(self, user_id: str) -> Optional[types.Relationship]:
"""
Fetches a Relationship instance from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if changes are made while
the instance exists the data will not be updated!
:param user_id: user-id of the Relationship
:return: The Relationship instance if it was found else None
"""
raw_data = self.find_relationship(user_id)
if raw_data:
return types.Relationship(raw_data, self)
else:
return None
get_room(self, room_id)
¶
Fetches a Room from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param room_id: id of the Room :return: The Room instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_room(self, room_id: str) -> Optional[types.TextRoom]:
"""
Fetches a Room from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if changes are made while
the instance exists the data will not be updated!
:param room_id: id of the Room
:return: The Room instance if it was found else None
"""
raw_data = self.find_room(room_id)
if raw_data:
return types.TextRoom(raw_data, self)
else:
return None
get_user(self, user_id)
¶
Fetches a User instance from the cache based on the passed id
The returned data of the instance is only a copy from the cache and if changes are made while the instance exists the data will not be updated!
:param user_id: id of the User :return: The User instance if it was found else None
Source code in openhivenpy\client\hivenclient.py
def get_user(self, user_id: str) -> Optional[types.User]:
"""
Fetches a User instance from the cache based on the passed id
---
The returned data of the instance is only a copy from the cache and if
changes are made while
the instance exists the data will not be updated!
:param user_id: id of the User
:return: The User instance if it was found else None
"""
raw_data = self.find_user(user_id)
if raw_data:
return types.User(raw_data, self)
else:
return None
run(self, token=None, *, loop=None, restart=False)
¶
Standard function for establishing a connection to Hiven
:param token: Token that should be used to connect to Hiven. If none is passed it will try to fetch the token using os.getenv() :param loop: Event loop that will be used to execute all async functions. Uses 'asyncio.get_event_loop()' to fetch the EventLoop. Will create a new one if no one was created yet. If the loop was passed during initialisation that one will be used if no loop is passed. If a new loop is passed, that one will be used for execution. :param restart: If set to True the Client will restart if an error is encountered!
Source code in openhivenpy\client\hivenclient.py
def run(
self,
token: str = None,
*,
loop: Optional[asyncio.AbstractEventLoop] = None,
restart: bool = False
) -> None:
"""
Standard function for establishing a connection to Hiven
:param token: Token that should be used to connect to Hiven. If none is
passed it will try to fetch the token using os.getenv()
:param loop: Event loop that will be used to execute all async
functions. Uses 'asyncio.get_event_loop()' to fetch the EventLoop.
Will create a new one if no one was created yet. If the loop was
passed during initialisation that one will be used if no loop is
passed. If a new loop is passed, that one will be used for execution.
:param restart: If set to True the Client will restart if an error is
encountered!
"""
try:
if token is None:
token = os.getenv('HIVEN_TOKEN')
if self._loop is not None:
self._loop = loop if loop is not None else self._loop
else:
try:
self._loop = loop if loop is not None else asyncio.get_event_loop()
except RuntimeError as e:
# If the function is called outside of the main thread a
# new event_loop must be created, so that the process can
# still be run. This will raise an exception though if the
# user tries to start the client while another loop already
# is running! Therefore run() should only be used when no
# event_loop was created yet that could interfere with the
# process, else connect() is available
if "There is no current event loop in thread" in str(e):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self._loop = asyncio.get_event_loop()
else:
raise
self.loop.run_until_complete(self.connect(token, restart=restart))
except KeyboardInterrupt:
pass
except (InvalidTokenError, WebSocketFailedError):
raise
except SessionCreateError:
raise
except Exception as e:
utils.log_traceback(
level='critical',
brief=f"Failed to keep alive connection to Hiven:",
exc_info=sys.exc_info()
)
raise HivenConnectionError(
"Failed to keep alive connection to Hiven") from e
Important
The HivenClient is already
inherited into this class (UserClient),
so all properties and methods can be used here as well.
openhivenpy.client.userclient.UserClient
¶
Class for the specific use of a user account on Hiven
Methods¶
block_user(self, user)
async
¶
Blocks a user on Hiven
:param user: Int or User Object used for the request :return: True if the request was successful :raises HTTPError: If the request was unsuccessful and an error was encountered
Source code in openhivenpy\client\userclient.py
async def block_user(self, user) -> bool:
"""
Blocks a user on Hiven
:param user: Int or User Object used for the request
:return: True if the request was successful
:raises HTTPError: If the request was unsuccessful and an error was
encountered
"""
try:
if type(user) is int:
user_id = str(user)
elif type(user) is types.User:
user_id = str(getattr(user, 'id'))
else:
raise TypeError(f"Expected User or int! Not {type(user)}")
resp = await self.http.put(endpoint=f"/relationships/@me/blocked/{user_id}")
return True
except Exception as e:
user_id = user if user is not None else getattr(user, 'id', None)
utils.log_traceback(
brief=f"Failed to block user with id {user_id}:",
exc_info=sys.exc_info()
)
raise
cancel_friend_request(self, user)
async
¶
Cancels an open friend request if it exists
:param user: Int or User Object used for the request :return: True if the request was successful :raises TypeError: If the passed user is not one of the types: int or types.User :raises HTTPError: If the request was unsuccessful and an error was encountered
Source code in openhivenpy\client\userclient.py
async def cancel_friend_request(self, user) -> bool:
"""
Cancels an open friend request if it exists
:param user: Int or User Object used for the request
:return: True if the request was successful
:raises TypeError: If the passed user is not one of the types: int or
types.User
:raises HTTPError: If the request was unsuccessful and an error was
encountered
"""
try:
if type(user) is int:
user_id = str(user)
elif type(user) is types.User:
user_id = str(getattr(user, 'id'))
else:
raise TypeError(
f"Expected openhivenpy.types.User or int! Not {type(user)}"
)
resp = await self.http.delete(
endpoint=f"/relationships/@me/friend-requests/{user_id}"
)
return True
except Exception as e:
user_id = user if user is not None else getattr(user, 'id', None)
utils.log_traceback(
brief=f"Failed to cancel the friend request of a user with id {user_id}:",
exc_info=sys.exc_info()
)
raise
fetch_current_friend_requests(self)
async
¶
Fetches all open friend requests on Hiven
:return: Returns a dict with all active friend requests if successful else it will raise an exception
Source code in openhivenpy\client\userclient.py
async def fetch_current_friend_requests(self) -> Union[dict, None]:
"""
Fetches all open friend requests on Hiven
:return: Returns a dict with all active friend requests if successful else it will raise an exception
"""
try:
resp = await self.http.get(endpoint=f"/relationships/@me/friend-requests")
resp = await resp.json()
data = resp.get('data')
incoming_ = data.get('incoming')
if incoming_:
data['incoming'] = []
for d in incoming_:
data['incoming'].append(types.LazyUser(d, self))
outgoing_ = data.get('outgoing')
if outgoing_:
data['outgoing'] = []
for d in outgoing_:
data['outgoing'].append(types.LazyUser(d, self))
return {
'incoming': data['incoming'],
'outgoing': data['outgoing']
}
except Exception as e:
utils.log_traceback(
brief="Failed to fetch the current open friend requests:",
exc_info=sys.exc_info()
)
raise
send_friend_request(self, user)
async
¶
Sends a friend request to a user
:param user: Int or User Object used for the request :return: True if the request was successful :raises HTTPError: If the request was unsuccessful and an error was encountered
Source code in openhivenpy\client\userclient.py
async def send_friend_request(self, user) -> bool:
"""
Sends a friend request to a user
:param user: Int or User Object used for the request
:return: True if the request was successful
:raises HTTPError: If the request was unsuccessful and an error was
encountered
"""
try:
if type(user) is int:
user_id = str(user)
elif type(user) is types.User:
user_id = str(getattr(user, 'id'))
else:
raise TypeError(f"Expected User or int! Not {type(user)}")
resp = await self.http.post(
endpoint=f"/relationships/@me/friend-requests", json={'user_id': f'{user_id}'}
)
if resp.status < 300:
return True
else:
raise HTTPFailedRequestError()
except Exception as e:
user_id = user if user is not None else getattr(user, 'id', None)
utils.log_traceback(
brief=f"Failed to send a friend request a user with id {user_id}:",
exc_info=sys.exc_info()
)
raise
unblock_user(self, user)
async
¶
Unblocks a user if the user is blocked
:param user: Int or User Object used for the request :return: True if the request was successful :raises HTTPError: If the request was unsuccessful and an error was encountered
Source code in openhivenpy\client\userclient.py
async def unblock_user(self, user) -> bool:
"""
Unblocks a user if the user is blocked
:param user: Int or User Object used for the request
:return: True if the request was successful
:raises HTTPError: If the request was unsuccessful and an error was
encountered
"""
try:
if type(user) is int:
user_id = str(user)
elif type(user) is types.User:
user_id = str(getattr(user, 'id'))
else:
raise TypeError(f"Expected User or int! Not {type(user)}")
resp = await self.http.delete(endpoint=f"/relationships/@me/blocked/{user_id}")
return True
except Exception as e:
user_id = user if user is not None else getattr(user, 'id', None)
utils.log_traceback(
brief=f"Failed to unblock a user with id {user_id}:",
exc_info=sys.exc_info()
)
raise
Important
The HivenClient is already
inherited into this class (BotClient),
so all properties and methods can be used here as well.
openhivenpy.client.botclient.BotClient
¶
Class for the specific use of a bot Application on Hiven