Skip to content

WebSocket Clients

openhivenpy.client.hivenclient.HivenClient

Main Class for connecting to Hiven and interacting with the API.

Attributes

account: Optional[str] property readonly

Returns the account id/string. Currently client-limited

api_version: Optional[str] property readonly

Returns the currently used Hiven API-version

application: Optional[bool] property readonly

Returns the application string passed. Currently client-limited

bio: Optional[str] property readonly

Bio of the user

blocked: Optional[bool] property readonly

Returns whether the user is blocked

bot: Optional[bool] property readonly

Returns true when the user is a bot

client_type: Optional[str] property readonly

Returns the Client-Type aka. the name of the class used

Possible options: HivenClient, UserClient, BotClient

client_user: Optional[openhivenpy.types.user.User] property readonly

The User Object of this client

close_timeout: Optional[int] property readonly

Set Close-Timeout, which if exceeded will cancel the connection

connection: Optional[openhivenpy.gateway.Connection] property readonly

Returns the Connection property

connection_status: Optional[int] property readonly

Returns the connection status as a string

email: Optional[str] property readonly

The e-mail of the user. Client-limited

email_verified: Optional[bool] property readonly

Returns True if the email is verified

flags: Union[int, str] property readonly

User flags represented as an numeric value/str

header: Optional[str] property readonly

The header of the user as a link

heartbeat: Optional[int] property readonly

Heartbeat in ms

host: Optional[str] property readonly

Returns the Hiven host url

house_ids: Optional[List[str]] property readonly

Returns the list of all the ids for all houses available from the cache

http: Optional[openhivenpy.gateway.http.HTTP] property readonly

Returns the HTTP Client from the Connection object if it exists

icon: Optional[str] property readonly

The icon of the user as a link

id: Optional[str] property readonly

Unique string id of the user

initialised: Optional[bool] property readonly

Returns whether the Client is initialised. This does not include the ready state though

location: Optional[str] property readonly

Set location of the user

log_websocket: Optional[str] property readonly

Returns whether the run configuration property log_websocket is enabled

loop: Optional[asyncio.events.AbstractEventLoop] property readonly

Returns the Asyncio Event-loop

message_broker: Optional[openhivenpy.gateway.messagebroker.MessageBroker] property readonly

Returns the Message-Broker handling all incoming events

mfa_enabled: Optional[bool] property readonly

Returns whether mfa is enabled

name: Optional[str] property readonly

Name of the user

open: Optional[bool] property readonly

Returns whether the Connection is open

presence: Optional[str] property readonly

Current presence of the User

queue_events: Optional[bool] property readonly

Returns whether the run configuration property queue_events is enabled

room_ids: Optional[List[str]] property readonly

Returns the list of all the ids for all rooms available from the cache.

This includes both house rooms and private rooms

startup_time: Optional[int] property readonly

Returns the amount of time it took for the bot to startup

storage: Optional[openhivenpy.client.cache.ClientCache] property readonly

Returns the Storage/Cache of the Client

token: Optional[str] property readonly

Returns the token of the Client

user_flags: Union[int, str] property readonly

Alias for flags

username: Optional[str] property readonly

Username of the user

website: Optional[str] property readonly

Set website of the user

Methods

__init__(self, token=None, *, loop=None, log_websocket=False, queue_events=False, host=None, api_version=None, heartbeat=None, close_timeout=None) special

Parameters:

Name Type Description Default
token str

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.

 None
loop AbstractEventLoop

Loop that will be used to run the Client. If a new one is passed on run() that one will be used instead

 None
log_websocket bool

If set to True will additionally log websocket messages and their content

 False
host Optional[str]

The host API endpoint of Hiven. Defaults to api.hiven.io

 None
api_version Optional[str]

The API version that should be used. Defaults to v1

 None
queue_events bool

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

 False
heartbeat Optional[int]

Intervals in which the bot will send heartbeats to the Websocket. Defaults to the pre-set environment variable heartbeat (default at 30000)

 None
close_timeout Optional[int]

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)

 None
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: str = token
    self._loop: asyncio.AbstractEventLoop = loop
    self._client_user: Optional[types.User] = None
    self._connection: Optional[Connection] = None
    self._storage: ClientCache = ClientCache(
        client=self,
        token=self._token
    )

    self._log_websocket: bool = log_websocket
    self._queue_events: bool = queue_events
    self._host: Optional[str] = host \
        if host is not None \
        else os.getenv("HIVEN_HOST")
    self._api_version: Optional[str] = api_version \
        if api_version is not None \
        else os.getenv("HIVEN_API_VERSION")
    self._heartbeat: Optional[int] = heartbeat \
        if heartbeat is not None \
        else int(os.getenv("WS_HEARTBEAT"))
    self._close_timeout: Optional[int] = close_timeout \
        if close_timeout is not None \
        else int(os.getenv("WS_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

Parameters:

Name Type Description Default
force bool

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

 False
remove_listeners bool

If set to True, it will remove all listeners including the ones created using @client.event(), add_multi_listener() and add_single_listener()

 True
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 set 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)
    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

Parameters:

Name Type Description Default
token str

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

 None
restart bool

If set to True the Client will restart if an error is encountered!

 False
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")

        self._connection = Connection(client=self)
        await self.connection.connect(restart=restart)

    except KeyboardInterrupt:
        pass

    except InvalidTokenError as e:
        raise e

    except Exception as e:
        try:
            await self.close()
        except Exception:
            ...  # ignoring, since this errors is too big for all of us

        del self._connection
        self._connection = None

        # Cleaning up the storage
        self.storage.closing_cleanup()

        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

Exceptions:

Type Description
HTTPError

If any HTTP error is raised while executing
Source code in openhivenpy\client\hivenclient.py 
async def edit(self, **kwargs) -> None:
    """
    Edits the Clients data on Hiven

    Available options: header, icon, bio, location, website, username

    :raise HTTPError: If any HTTP error is raised while executing
    """
    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)}
                )
            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 e

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

Parameters:

Name Type Description Default
entity_id str

id of the Entity

 required

Returns:

Type Description
Optional[dict]

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

Parameters:

Name Type Description Default
house_id str

id of the House

 required

Returns:

Type Description
Optional[dict]

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_house_member(self, member_id, house_id)

Fetches the raw data of a member

Parameters:

Name Type Description Default
member_id str

The id of the Member which should be fetched

 required
house_id str

The id of the House the Member is in

 required

Returns:

Type Description
Optional[dict]

The dictionary of the member if it was found
Source code in openhivenpy\client\hivenclient.py 
def find_house_member(
        self, member_id: str, house_id: str
) -> Optional[dict]:
    """
    Fetches the raw data of a member

    :param member_id: The id of the Member which should be fetched
    :param house_id: The id of the House the Member is in
    :return: The dictionary of the member if it was found
    """
    return self.storage['houses']\
        .get(house_id, {})\
        .get('members', {})\
        .get(member_id)

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

Parameters:

Name Type Description Default
room_id str

id of the PrivateGroupRoom

 required

Returns:

Type Description
Optional[dict]

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

Parameters:

Name Type Description Default
room_id str

id of the PrivateRoom

 required

Returns:

Type Description
Optional[dict]

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

Parameters:

Name Type Description Default
user_id str

user-id of the Relationship

 required

Returns:

Type Description
Optional[dict]

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

Parameters:

Name Type Description Default
room_id str

id of the Room

 required

Returns:

Type Description
Optional[dict]

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

Parameters:

Name Type Description Default
user_id str

id of the User

 required

Returns:

Type Description
Optional[dict]

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!

Parameters:

Name Type Description Default
entity_id str

id of the Entity

 required

Returns:

Type Description
Optional[openhivenpy.types.entity.Entity]

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!

Parameters:

Name Type Description Default
house_id str

id of the House

 required

Returns:

Type Description
Optional[openhivenpy.types.house.House]

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_house_member(self, member_id, house_id)

Fetches a member from the cache based on the id

Parameters:

Name Type Description Default
member_id str

The id of the Member which should be fetched

 required
house_id str

The id of the House the Member is in

 required

Returns:

Type Description
Optional[openhivenpy.types.member.Member]

The Member Instance if it exists else returns None
Source code in openhivenpy\client\hivenclient.py 
def get_house_member(
        self, member_id: str, house_id: str
) -> Optional[types.Member]:
    """
    Fetches a member from the cache based on the id

    :param member_id: The id of the Member which should be fetched
    :param house_id: The id of the House the Member is in
    :return: The Member Instance if it exists else returns None
    """
    cached_member = self.find_house_member(member_id, house_id)
    if cached_member:
        return types.Member(cached_member, self._client)

    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!

Parameters:

Name Type Description Default
room_id str

id of the PrivateGroupRoom

 required

Returns:

Type Description
Optional[openhivenpy.types.private_room.PrivateGroupRoom]

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!

Parameters:

Name Type Description Default
room_id str

id of the PrivateRoom

 required

Returns:

Type Description
Optional[openhivenpy.types.private_room.PrivateRoom]

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!

Parameters:

Name Type Description Default
user_id str

user-id of the Relationship

 required

Returns:

Type Description
Optional[openhivenpy.types.relationship.Relationship]

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!

Parameters:

Name Type Description Default
room_id str

id of the Room

 required

Returns:

Type Description
Optional[openhivenpy.types.textroom.TextRoom]

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!

Parameters:

Name Type Description Default
user_id str

id of the User

 required

Returns:

Type Description
Optional[openhivenpy.types.user.User]

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

Parameters:

Name Type Description Default
token str

Token that should be used to connect to Hiven. If none is passed it will try to fetch the token using os.getenv()

 None
loop Optional[asyncio.events.AbstractEventLoop]

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.

 None
restart bool

If set to True the Client will restart if an error is encountered!

 False
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!
    """
    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 e

    self.loop.run_until_complete(self.connect(token, restart=restart))

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

Parameters:

Name Type Description Default
user

Int or User Object used for the request

 required

Exceptions:

Type Description
HTTPError

If any HTTP error is raised while executing
Source code in openhivenpy\client\userclient.py 
async def block_user(self, user) -> None:
    """
    Blocks a user on Hiven

    :param user: Int or User Object used for the request
    :raise HTTPError: If any HTTP error is raised while executing
    """
    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)}")

        await self.http.put(f"/relationships/@me/blocked/{user_id}")

    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 e

cancel_friend_request(self, user) async

Cancels an open friend request if it exists

Parameters:

Name Type Description Default
user

Int or User Object used for the request

 required

Exceptions:

Type Description
TypeError

If the passed user is not one of the types: int or types.User
HTTPError

If any HTTP error is raised while executing
Source code in openhivenpy\client\userclient.py 
async def cancel_friend_request(self, user) -> None:
    """
    Cancels an open friend request if it exists

    :param user: Int or User Object used for the request
    :raises TypeError: If the passed user is not one of the types: int or
     types.User
    :raise HTTPError: If any HTTP error is raised while executing
    """
    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)}"
            )

        await self.http.delete(
            f"/relationships/@me/friend-requests/{user_id}"
        )

    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 e

fetch_current_friend_requests(self) async

Fetches all open friend requests on Hiven

Returns:

Type Description
Optional[dict]

Returns a dict with all active friend requests if successful

Exceptions:

Type Description
HTTPError

If any HTTP error is raised while executing
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
    :raise HTTPError: If any HTTP error is raised while executing
    """
    try:
        resp = await self.http.get(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 e

send_friend_request(self, user) async

Sends a friend request to a user

Parameters:

Name Type Description Default
user

Int or User Object used for the request

 required

Exceptions:

Type Description
HTTPError

If any HTTP error is raised while executing
Source code in openhivenpy\client\userclient.py 
async def send_friend_request(self, user) -> None:
    """
    Sends a friend request to a user

    :param user: Int or User Object used for the request
    :raise HTTPError: If any HTTP error is raised while executing
    """
    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)}")

        await self.http.post(
            endpoint=f"/relationships/@me/friend-requests",
            json={'user_id': user_id}
        )

    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 e

unblock_user(self, user) async

Unblocks a user if the user is blocked

Parameters:

Name Type Description Default
user

Int or User Object used for the request

 required

Exceptions:

Type Description
HTTPError

If any HTTP error is raised while executing
Source code in openhivenpy\client\userclient.py 
async def unblock_user(self, user) -> None:
    """
    Unblocks a user if the user is blocked

    :param user: Int or User Object used for the request
    :raise HTTPError: If any HTTP error is raised while executing
    """
    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)}")

        await self.http.delete(
            f"/relationships/@me/blocked/{user_id}"
        )

    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 e

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

Last update: 2021-08-22