Replace trigger_typing with awaiting typing context manager

This commit is contained in:
NextChai 2022-05-02 04:36:18 -04:00 committed by GitHub
parent 9deac84eac
commit 300fb0411e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 38 additions and 25 deletions

View File

@ -1544,32 +1544,30 @@ class Messageable:
await ret.delete(delay=delete_after)
return ret
async def trigger_typing(self) -> None:
"""|coro|
Triggers a *typing* indicator to the destination.
*Typing* indicator will go away after 10 seconds, or after a message is sent.
"""
channel = await self._get_channel()
await self._state.http.send_typing(channel.id)
def typing(self) -> Typing:
"""Returns an asynchronous context manager that allows you to type for an indefinite period of time.
This is useful for denoting long computations in your bot.
"""Returns an asynchronous context manager that allows you to send a typing indicator to
the destination for an indefinite period of time, or 10 seconds if the context manager
is called using ``await``.
Example Usage: ::
async with channel.typing():
# simulate something heavy
await asyncio.sleep(10)
await asyncio.sleep(20)
await channel.send('done!')
await channel.send('Done!')
Example Usage: ::
await channel.typing()
# Do some computational magic for about 10 seconds
await channel.send('Done!')
.. versionchanged:: 2.0
This no longer works with the ``with`` syntax, ``async with`` must be used instead.
.. versionchanged:: 2.0
Added functionality to ``await`` the context manager to send a typing indicator for 10 seconds.
"""
return Typing(self)

View File

@ -25,10 +25,10 @@ DEALINGS IN THE SOFTWARE.
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING, Optional, Type, TypeVar
from typing import TYPE_CHECKING, Generator, Optional, Type, TypeVar
if TYPE_CHECKING:
from .abc import Messageable
from .abc import Messageable, MessageableChannel
from types import TracebackType
@ -53,21 +53,32 @@ class Typing:
def __init__(self, messageable: Messageable) -> None:
self.loop: asyncio.AbstractEventLoop = messageable._state.loop
self.messageable: Messageable = messageable
self.channel: Optional[MessageableChannel] = None
async def _get_channel(self) -> MessageableChannel:
if self.channel:
return self.channel
self.channel = channel = await self.messageable._get_channel()
return channel
async def wrapped_typer(self) -> None:
channel = await self._get_channel()
await channel._state.http.send_typing(channel.id)
def __await__(self) -> Generator[None, None, None]:
return self.wrapped_typer().__await__()
async def do_typing(self) -> None:
try:
channel = self._channel
except AttributeError:
channel = await self.messageable._get_channel()
channel = await self._get_channel()
typing = channel._state.http.send_typing
while True:
await typing(channel.id)
await asyncio.sleep(5)
await typing(channel.id)
async def __aenter__(self) -> None:
self._channel = channel = await self.messageable._get_channel()
channel = await self._get_channel()
await channel._state.http.send_typing(channel.id)
self.task: asyncio.Task[None] = self.loop.create_task(self.do_typing())
self.task.add_done_callback(_typing_done_callback)

View File

@ -1152,6 +1152,10 @@ The following have been removed:
- Consider using the newer documented :func:`on_socket_event_type` event instead.
- ``abc.Messageable.trigger_typing``
- Use :meth:`abc.Messageable.typing` with ``await`` instead.
Miscellaneous Changes
----------------------