Connection state
attrs
class
ConnectionState
¶
Attr attributes:
Name | Type | Description |
---|---|---|
client |
Client |
The bot's client |
intents |
Intents |
The event intents in use |
shard_id |
int |
The shard ID of this state |
gateway |
Union[naff.api.gateway.gateway.GatewayClient, naff.client.const.Missing] |
The websocket connection for the Discord Gateway. |
start_time |
Union[datetime.datetime, naff.client.const.Missing] |
The DateTime the bot started at |
gateway_url |
str |
The URL that the gateway should connect to. |
gateway_started |
Event |
Event to check if the gateway has been started. |
Source code in naff/api/gateway/state.py
@define(kw_only=False)
class ConnectionState:
client: "Client"
"""The bot's client"""
intents: Intents
"""The event intents in use"""
shard_id: int
"""The shard ID of this state"""
_shard_ready: asyncio.Event = field(default=None)
"""Indicates that this state is now ready"""
gateway: Absent[GatewayClient] = MISSING
"""The websocket connection for the Discord Gateway."""
start_time: Absent[datetime] = MISSING
"""The DateTime the bot started at"""
gateway_url: str = MISSING
"""The URL that the gateway should connect to."""
gateway_started: asyncio.Event = asyncio.Event()
"""Event to check if the gateway has been started."""
_shard_task: asyncio.Task | None = None
def __attrs_post_init__(self, *args, **kwargs) -> None:
self._shard_ready = asyncio.Event()
@property
def latency(self) -> float:
"""Returns the latency of the websocket connection."""
return self.gateway.average_latency
@property
def average_latency(self) -> float:
"""Returns the average latency of the websocket connection."""
return self.gateway.average_latency
@property
def presence(self) -> dict:
"""Returns the presence of the bot."""
return {
"status": self.client._status,
"activities": [self.client._activity.to_dict()] if self.client._activity else [],
}
async def start(self) -> None:
"""Connect to the Discord Gateway."""
self.gateway_url = await self.client.http.get_gateway()
logger.debug(f"Starting Shard ID {self.shard_id}")
self.start_time = datetime.now()
self._shard_task = asyncio.create_task(self._ws_connect())
self.gateway_started.set()
# Historically this method didn't return until the connection closed
# so we need to wait for the task to exit.
await self._shard_task
async def stop(self) -> None:
"""Disconnect from the Discord Gateway."""
logger.debug(f"Shutting down shard ID {self.shard_id}")
if self.gateway is not None:
self.gateway.close()
self.gateway = None
if self._shard_task is not None:
await self._shard_task
self._shard_task = None
self.gateway_started.clear()
def clear_ready(self) -> None:
"""Clear the ready event."""
self._shard_ready.clear()
self.client._ready.clear() # noinspection PyProtectedMember
async def _ws_connect(self) -> None:
"""Connect to the Discord Gateway."""
logger.info(f"Shard {self.shard_id} is attempting to connect to gateway...")
try:
async with GatewayClient(self, (self.shard_id, self.client.total_shards)) as self.gateway:
try:
await self.gateway.run()
finally:
self._shard_ready.clear()
if self.client.total_shards == 1:
self.client.dispatch(events.Disconnect())
else:
self.client.dispatch(events.ShardDisconnect(self.shard_id))
except WebSocketClosed as ex:
if ex.code == 4011:
raise NaffException("Your bot is too large, you must use shards") from None
elif ex.code == 4013:
raise NaffException(f"Invalid Intents have been passed: {self.intents}") from None
elif ex.code == 4014:
raise NaffException(
"You have requested privileged intents that have not been enabled or approved. Check the developer dashboard"
) from None
raise
except Exception as e:
self.client.dispatch(events.Disconnect())
logger.error("".join(traceback.format_exception(type(e), e, e.__traceback__)))
async def change_presence(
self, status: Optional[Union[str, Status]] = Status.ONLINE, activity: Absent[Union[Activity, str]] = MISSING
) -> None:
"""
Change the bots presence.
Args:
status: The status for the bot to be. i.e. online, afk, etc.
activity: The activity for the bot to be displayed as doing.
note::
Bots may only be `playing` `streaming` `listening` `watching` or `competing`, other activity types are likely to fail.
"""
if activity is not MISSING:
if activity is None:
activity = []
else:
if not isinstance(activity, Activity):
# squash whatever the user passed into an activity
activity = Activity.create(name=str(activity))
if activity.type == ActivityType.STREAMING:
if not activity.url:
logger.warning("Streaming activity cannot be set without a valid URL attribute")
elif activity.type not in [
ActivityType.GAME,
ActivityType.STREAMING,
ActivityType.LISTENING,
ActivityType.WATCHING,
ActivityType.COMPETING,
]:
logger.warning(f"Activity type `{ActivityType(activity.type).name}` may not be enabled for bots")
else:
activity = self.client.activity
if status:
if not isinstance(status, Status):
try:
status = Status[status.upper()]
except KeyError:
raise ValueError(f"`{status}` is not a valid status type. Please use the Status enum") from None
else:
# in case the user set status to None
if self.client.status:
status = self.client.status
else:
logger.warning("Status must be set to a valid status type, defaulting to online")
status = Status.ONLINE
self.client._status = status
self.client._activity = activity
await self.gateway.change_presence(activity.to_dict() if activity else None, status)
def get_voice_state(self, guild_id: "Snowflake_Type") -> Optional["naff.ActiveVoiceState"]:
"""
Get the bot's voice state for a guild.
Args:
guild_id: The target guild's id.
Returns:
The bot's voice state for the guild if connected, otherwise None.
"""
return self.client.cache.get_bot_voice_state(guild_id)
async def voice_connect(
self, guild_id, channel_id, muted: bool = False, deafened: bool = False
) -> "naff.ActiveVoiceState":
"""
Connect to a voice channel.
Args:
guild_id: id of the guild the voice channel is in.
channel_id: id of the voice channel client wants to join.
muted: Whether the bot should be muted when connected.
deafened: Whether the bot should be deafened when connected.
Returns:
The new active voice state on successfully connection.
"""
voice_state = naff.ActiveVoiceState(
client=self.client, guild_id=guild_id, channel_id=channel_id, self_mute=muted, self_deaf=deafened
)
await voice_state.connect()
self.client.cache.place_bot_voice_state(voice_state)
return voice_state
property
readonly
latency: float
¶
Returns the latency of the websocket connection.
property
readonly
average_latency: float
¶
Returns the average latency of the websocket connection.
property
readonly
presence: dict
¶
Returns the presence of the bot.
async
method
start(self)
¶
Connect to the Discord Gateway.
Source code in naff/api/gateway/state.py
async def start(self) -> None:
"""Connect to the Discord Gateway."""
self.gateway_url = await self.client.http.get_gateway()
logger.debug(f"Starting Shard ID {self.shard_id}")
self.start_time = datetime.now()
self._shard_task = asyncio.create_task(self._ws_connect())
self.gateway_started.set()
# Historically this method didn't return until the connection closed
# so we need to wait for the task to exit.
await self._shard_task
async
method
stop(self)
¶
Disconnect from the Discord Gateway.
Source code in naff/api/gateway/state.py
async def stop(self) -> None:
"""Disconnect from the Discord Gateway."""
logger.debug(f"Shutting down shard ID {self.shard_id}")
if self.gateway is not None:
self.gateway.close()
self.gateway = None
if self._shard_task is not None:
await self._shard_task
self._shard_task = None
self.gateway_started.clear()
method
clear_ready(self)
¶
Clear the ready event.
Source code in naff/api/gateway/state.py
def clear_ready(self) -> None:
"""Clear the ready event."""
self._shard_ready.clear()
self.client._ready.clear() # noinspection PyProtectedMember
async
method
change_presence(self, status, activity)
¶
Change the bots presence.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
status |
Union[str, naff.models.discord.enums.Status] |
The status for the bot to be. i.e. online, afk, etc. |
<Status.ONLINE: 'online'> |
activity |
Union[naff.models.discord.activity.Activity, str, naff.client.const.Missing] |
The activity for the bot to be displayed as doing. |
Missing |
note::
Bots may only be playing
streaming
listening
watching
or competing
, other activity types are likely to fail.
Source code in naff/api/gateway/state.py
async def change_presence(
self, status: Optional[Union[str, Status]] = Status.ONLINE, activity: Absent[Union[Activity, str]] = MISSING
) -> None:
"""
Change the bots presence.
Args:
status: The status for the bot to be. i.e. online, afk, etc.
activity: The activity for the bot to be displayed as doing.
note::
Bots may only be `playing` `streaming` `listening` `watching` or `competing`, other activity types are likely to fail.
"""
if activity is not MISSING:
if activity is None:
activity = []
else:
if not isinstance(activity, Activity):
# squash whatever the user passed into an activity
activity = Activity.create(name=str(activity))
if activity.type == ActivityType.STREAMING:
if not activity.url:
logger.warning("Streaming activity cannot be set without a valid URL attribute")
elif activity.type not in [
ActivityType.GAME,
ActivityType.STREAMING,
ActivityType.LISTENING,
ActivityType.WATCHING,
ActivityType.COMPETING,
]:
logger.warning(f"Activity type `{ActivityType(activity.type).name}` may not be enabled for bots")
else:
activity = self.client.activity
if status:
if not isinstance(status, Status):
try:
status = Status[status.upper()]
except KeyError:
raise ValueError(f"`{status}` is not a valid status type. Please use the Status enum") from None
else:
# in case the user set status to None
if self.client.status:
status = self.client.status
else:
logger.warning("Status must be set to a valid status type, defaulting to online")
status = Status.ONLINE
self.client._status = status
self.client._activity = activity
await self.gateway.change_presence(activity.to_dict() if activity else None, status)
method
get_voice_state(self, guild_id)
¶
Get the bot's voice state for a guild.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id |
Snowflake_Type |
The target guild's id. |
required |
Returns:
Type | Description |
---|---|
Optional[naff.ActiveVoiceState] |
The bot's voice state for the guild if connected, otherwise None. |
Source code in naff/api/gateway/state.py
def get_voice_state(self, guild_id: "Snowflake_Type") -> Optional["naff.ActiveVoiceState"]:
"""
Get the bot's voice state for a guild.
Args:
guild_id: The target guild's id.
Returns:
The bot's voice state for the guild if connected, otherwise None.
"""
return self.client.cache.get_bot_voice_state(guild_id)
async
method
voice_connect(self, guild_id, channel_id, muted, deafened)
¶
Connect to a voice channel.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
guild_id |
id of the guild the voice channel is in. |
required | |
channel_id |
id of the voice channel client wants to join. |
required | |
muted |
bool |
Whether the bot should be muted when connected. |
False |
deafened |
bool |
Whether the bot should be deafened when connected. |
False |
Returns:
Type | Description |
---|---|
naff.ActiveVoiceState |
The new active voice state on successfully connection. |
Source code in naff/api/gateway/state.py
async def voice_connect(
self, guild_id, channel_id, muted: bool = False, deafened: bool = False
) -> "naff.ActiveVoiceState":
"""
Connect to a voice channel.
Args:
guild_id: id of the guild the voice channel is in.
channel_id: id of the voice channel client wants to join.
muted: Whether the bot should be muted when connected.
deafened: Whether the bot should be deafened when connected.
Returns:
The new active voice state on successfully connection.
"""
voice_state = naff.ActiveVoiceState(
client=self.client, guild_id=guild_id, channel_id=channel_id, self_mute=muted, self_deaf=deafened
)
await voice_state.connect()
self.client.cache.place_bot_voice_state(voice_state)
return voice_state