mirrors/discord.py
1
0
Fork 0
mirror of https://github.com/Rapptz/discord.py.git synced 2025-06-07 20:28:38 +00:00
Code Issues Projects Releases Wiki Activity

Typehint context_managers.py

Browse Source
This commit is contained in:
Stocker 2021-08-20 19:50:39 -04:00 committed by GitHub
parent b6590d7f56
commit ef32f6d882
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 28 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
discord/context_managers.py
View File

@ -22,13 +22,23 @@ FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE. DEALINGS IN THE SOFTWARE.
""" """
from __future__ import annotations
import asyncio import asyncio
from typing import TYPE_CHECKING, TypeVar, Optional, Type
if TYPE_CHECKING:
from .abc import Messageable
from types import TracebackType
TypingT = TypeVar('TypingT', bound='Typing')
__all__ = ( __all__ = (
'Typing', 'Typing',
) )
def _typing_done_callback(fut): def _typing_done_callback(fut: asyncio.Future) -> None:
# just retrieve any exception and call it a day # just retrieve any exception and call it a day
try: try:
fut.exception() fut.exception()
@ -36,11 +46,11 @@ def _typing_done_callback(fut):
pass pass
class Typing: class Typing:
def __init__(self, messageable): def __init__(self, messageable: Messageable) -> None:
self.loop = messageable._state.loop self.loop: asyncio.AbstractEventLoop = messageable._state.loop
self.messageable = messageable self.messageable: Messageable = messageable
async def do_typing(self): async def do_typing(self) -> None:
try: try:
channel = self._channel channel = self._channel
except AttributeError: except AttributeError:
@ -52,18 +62,26 @@ class Typing:
await typing(channel.id) await typing(channel.id)
await asyncio.sleep(5) await asyncio.sleep(5)
def __enter__(self): def __enter__(self: TypingT) -> TypingT:
self.task = asyncio.ensure_future(self.do_typing(), loop=self.loop) self.task: asyncio.Task = self.loop.create_task(self.do_typing())
self.task.add_done_callback(_typing_done_callback) self.task.add_done_callback(_typing_done_callback)
return self return self
def __exit__(self, exc_type, exc, tb): def __exit__(self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
self.task.cancel() self.task.cancel()
async def __aenter__(self): async def __aenter__(self: TypingT) -> TypingT:
self._channel = channel = await self.messageable._get_channel() self._channel = channel = await self.messageable._get_channel()
await channel._state.http.send_typing(channel.id) await channel._state.http.send_typing(channel.id)
return self.__enter__() return self.__enter__()
async def __aexit__(self, exc_type, exc, tb): async def __aexit__(self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
self.task.cancel() self.task.cancel()