auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-05-09 04:08:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Overload signatures

Browse Source
This commit is contained in:
Josh Wu
2023-09-21 02:04:30 +08:00
committed by Lucas Abel
parent eb9d5632bc
commit 3b14078646
1 changed files with 38 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
bumble/utils.py
View File

@@ -21,7 +21,18 @@ import logging
import traceback import traceback
import collections import collections
import sys import sys
from typing import Awaitable, Set, TypeVar, List, Tuple, Callable, Any, Optional, Union from typing import (
Awaitable,
Set,
TypeVar,
List,
Tuple,
Callable,
Any,
Optional,
Union,
overload,
)
from functools import wraps from functools import wraps
from pyee import EventEmitter from pyee import EventEmitter
@@ -65,13 +76,15 @@ def composite_listener(cls):
return cls return cls
# -----------------------------------------------------------------------------
_Handler = TypeVar('_Handler', bound=Callable) _Handler = TypeVar('_Handler', bound=Callable)
# -----------------------------------------------------------------------------
class EventWatcher: class EventWatcher:
'''A wrapper class to control the lifecycle of event handlers better. '''A wrapper class to control the lifecycle of event handlers better.
Usage: Usage:
```
watcher = EventWatcher() watcher = EventWatcher()
def on_foo(): def on_foo():
@@ -84,13 +97,16 @@ class EventWatcher:
# Close all event handlers watching through this watcher # Close all event handlers watching through this watcher
watcher.close() watcher.close()
```
As context: As context:
```
with contextlib.closing(EventWatcher()) as context: with contextlib.closing(EventWatcher()) as context:
@context.on(emitter, 'foo') @context.on(emitter, 'foo')
def on_foo(): def on_foo():
... ...
# on_foo() has been removed here! # on_foo() has been removed here!
```
''' '''
handlers: List[Tuple[EventEmitter, str, Callable[..., Any]]] handlers: List[Tuple[EventEmitter, str, Callable[..., Any]]]
@@ -98,6 +114,14 @@ class EventWatcher:
def __init__(self) -> None: def __init__(self) -> None:
self.handlers = [] self.handlers = []
@overload
def on(self, emitter: EventEmitter, event: str) -> Callable[[_Handler], _Handler]:
...
@overload
def on(self, emitter: EventEmitter, event: str, handler: _Handler) -> _Handler:
...
def on( def on(
self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None
) -> Union[_Handler, Callable[[_Handler], _Handler]]: ) -> Union[_Handler, Callable[[_Handler], _Handler]]:
@@ -109,12 +133,21 @@ class EventWatcher:
handler: (Optional) Event handler. When nothing passed, this method works as a decorator. handler: (Optional) Event handler. When nothing passed, this method works as a decorator.
''' '''
def wrapper(f: _Handler): def wrapper(f: _Handler) -> _Handler:
self.handlers.append((emitter, event, f)) self.handlers.append((emitter, event, f))
emitter.on(event, f) emitter.on(event, f)
return f
return wrapper if handler is None else wrapper(handler) return wrapper if handler is None else wrapper(handler)
@overload
def once(self, emitter: EventEmitter, event: str) -> Callable[[_Handler], _Handler]:
...
@overload
def once(self, emitter: EventEmitter, event: str, handler: _Handler) -> _Handler:
...
def once( def once(
self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None self, emitter: EventEmitter, event: str, handler: Optional[_Handler] = None
) -> Union[_Handler, Callable[[_Handler], _Handler]]: ) -> Union[_Handler, Callable[[_Handler], _Handler]]:
@@ -126,9 +159,10 @@ class EventWatcher:
handler: (Optional) Event handler. When nothing passed, this method works as a decorator. handler: (Optional) Event handler. When nothing passed, this method works as a decorator.
''' '''
def wrapper(f: _Handler): def wrapper(f: _Handler) -> _Handler:
self.handlers.append((emitter, event, f)) self.handlers.append((emitter, event, f))
emitter.once(event, f) emitter.once(event, f)
return f
return wrapper if handler is None else wrapper(handler) return wrapper if handler is None else wrapper(handler)