mirror of
https://github.com/python/cpython.git
synced 2025-11-04 07:31:38 +00:00
b9127dd6ee
* Update code after merge review from 1st1 * Use a sentinel approach for loop parameter Remove unnecessary _get_running_loop patching * Use more clear function name (_verify_parameter_is_marker -> _verify_no_loop) * Add init method to _LoopBoundMixin to check that loop param wasn't used
31 lines
803 B
Python
31 lines
803 B
Python
"""Event loop mixins."""
|
|
|
|
import threading
|
|
from . import events
|
|
|
|
_global_lock = threading.Lock()
|
|
|
|
# Used as a sentinel for loop parameter
|
|
_marker = object()
|
|
|
|
|
|
class _LoopBoundMixin:
|
|
_loop = None
|
|
|
|
def __init__(self, *, loop=_marker):
|
|
if loop is not _marker:
|
|
raise TypeError(
|
|
f'As of 3.10, the *loop* parameter was removed from '
|
|
f'{type(self).__name__}() since it is no longer necessary'
|
|
)
|
|
|
|
def _get_loop(self):
|
|
loop = events._get_running_loop()
|
|
|
|
if self._loop is None:
|
|
with _global_lock:
|
|
if self._loop is None:
|
|
self._loop = loop
|
|
if loop is not self._loop:
|
|
raise RuntimeError(f'{self!r} is bound to a different event loop')
|
|
return loop
|