from .log import logger

__all__ = ['BaseEventLoop']

__all__ = 'BaseEventLoop',

# Minimum number of _scheduled timer handles before cleanup of

def _run_until_complete_cb(fut):

exc = fut._exception

if (isinstance(exc, BaseException)

and not isinstance(exc, Exception)):

if isinstance(exc, BaseException) and not isinstance(exc, Exception):

# Issue #22429: run_forever() already finished, no need to

# stop it.

return

self._waiters = []

def __repr__(self):

return '<%s sockets=%r>' % (self.__class__.__name__, self.sockets)

return f'<{self.__class__.__name__} sockets={self.sockets!r}>'

def _attach(self):

assert self.sockets is not None

self._asyncgens_shutdown_called = False

def __repr__(self):

return ('<%s running=%s closed=%s debug=%s>'

% (self.__class__.__name__, self.is_running(),

self.is_closed(), self.get_debug()))

return (

f'<{self.__class__.__name__} running={self.is_running()} '

f'closed={self.is_closed()} debug={self.get_debug()}>'

)

def create_future(self):

"""Create a Future object attached to the loop."""

def _asyncgen_firstiter_hook(self, agen):

if self._asyncgens_shutdown_called:

warnings.warn(

"asynchronous generator {!r} was scheduled after "

"loop.shutdown_asyncgens() call".format(agen),

f"asynchronous generator {agen!r} was scheduled after "

f"loop.shutdown_asyncgens() call",

ResourceWarning, source=self)

self._asyncgens.add(agen)

for result, agen in zip(results, closing_agens):

if isinstance(result, Exception):

self.call_exception_handler({

'message': 'an error occurred during closing of '

'asynchronous generator {!r}'.format(agen),

'message': f'an error occurred during closing of '

f'asynchronous generator {agen!r}',

'exception': result,

'asyncgen': agen

})

def __del__(self):

if not self.is_closed():

warnings.warn("unclosed event loop %r" % self, ResourceWarning,

warnings.warn(f"unclosed event loop {self!r}", ResourceWarning,

source=self)

if not self.is_running():

self.close()

if (coroutines.iscoroutine(callback) or

coroutines.iscoroutinefunction(callback)):

raise TypeError(

"coroutines cannot be used with {}()".format(method))

f"coroutines cannot be used with {method}()")

if not callable(callback):

raise TypeError(

'a callable object was expected by {}(), got {!r}'.format(

method, callback))

f'a callable object was expected by {method}(), '

f'got {callback!r}')

def _call_soon(self, callback, args):

handle = events.Handle(callback, args, self)

self._default_executor = executor

def _getaddrinfo_debug(self, host, port, family, type, proto, flags):

msg = ["%s:%r" % (host, port)]

msg = [f"{host}:{port!r}"]

if family:

msg.append('family=%r' % family)

msg.append(f'family={family!r}' % family)

if type:

msg.append('type=%r' % type)

msg.append(f'type={type!r}')

if proto:

msg.append('proto=%r' % proto)

msg.append(f'proto={proto!r}')

if flags:

msg.append('flags=%r' % flags)

msg.append(f'flags={flags!r}')

msg = ', '.join(msg)

logger.debug('Get address info %s', msg)

t0 = self.time()

addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags)

dt = self.time() - t0

msg = ('Getting address info %s took %.3f ms: %r'

% (msg, dt * 1e3, addrinfo))

msg = f'Getting address info {msg} took {dt * 1e3:.3f}ms: {addrinfo!r}'

if dt >= self.slow_callback_duration:

logger.info(msg)

else:

sock.bind(laddr)

break

except OSError as exc:

exc = OSError(

exc.errno, 'error while '

'attempting to bind on address '

'{!r}: {}'.format(

laddr, exc.strerror.lower()))

msg = (

f'error while attempting to bind on '

f'address {laddr!r}: '

f'{exc.strerror.lower()}'

)

exc = OSError(exc.errno, msg)

exceptions.append(exc)

else:

sock.close()

# Disallowing AF_UNIX in this method, breaks backwards

# compatibility.

raise ValueError(

'A Stream Socket was expected, got {!r}'.format(sock))

f'A Stream Socket was expected, got {sock!r}')

transport, protocol = await self._create_connection_transport(

sock, protocol_factory, ssl, server_hostname)

if sock is not None:

if not _is_dgram_socket(sock):

raise ValueError(

'A UDP Socket was expected, got {!r}'.format(sock))

f'A UDP Socket was expected, got {sock!r}')

if (local_addr or remote_addr or

family or proto or flags or

reuse_address or reuse_port or allow_broadcast):

family=family, proto=proto, flags=flags,

reuse_address=reuse_address, reuse_port=reuse_port,

allow_broadcast=allow_broadcast)

problems = ', '.join(

'{}={}'.format(k, v) for k, v in opts.items() if v)

problems = ', '.join(f'{k}={v}' for k, v in opts.items() if v)

raise ValueError(

'socket modifier keyword arguments can not be used '

'when sock is specified. ({})'.format(problems))

f'socket modifier keyword arguments can not be used '

f'when sock is specified. ({problems})')

sock.setblocking(False)

r_addr = None

else:

type=socket.SOCK_STREAM,

flags=flags, loop=self)

if not infos:

raise OSError('getaddrinfo({!r}) returned empty list'.format(host))

raise OSError(f'getaddrinfo({host!r}) returned empty list')

return infos

async def create_server(self, protocol_factory, host=None, port=None,

reuse_port=None):

"""Create a TCP server.

The host parameter can be a string, in that case the TCP server is bound

to host and port.

The host parameter can be a string, in that case the TCP server is

bound to host and port.

The host parameter can also be a sequence of strings and in that case

the TCP server is bound to all hosts of the sequence. If a host

if sock is None:

raise ValueError('Neither host/port nor sock were specified')

if not _is_stream_socket(sock):

raise ValueError(

'A Stream Socket was expected, got {!r}'.format(sock))

raise ValueError(f'A Stream Socket was expected, got {sock!r}')

sockets = [sock]

server = Server(self, sockets)

returns a (transport, protocol) pair.

"""

if not _is_stream_socket(sock):

raise ValueError(

'A Stream Socket was expected, got {!r}'.format(sock))

raise ValueError(f'A Stream Socket was expected, got {sock!r}')

transport, protocol = await self._create_connection_transport(

sock, protocol_factory, ssl, '', server_side=True)

def _log_subprocess(self, msg, stdin, stdout, stderr):

info = [msg]

if stdin is not None:

info.append('stdin=%s' % _format_pipe(stdin))

info.append(f'stdin={_format_pipe(stdin)}')

if stdout is not None and stderr == subprocess.STDOUT:

info.append('stdout=stderr=%s' % _format_pipe(stdout))

info.append(f'stdout=stderr={_format_pipe(stdout)}')

else:

if stdout is not None:

info.append('stdout=%s' % _format_pipe(stdout))

info.append(f'stdout={_format_pipe(stdout)}')

if stderr is not None:

info.append('stderr=%s' % _format_pipe(stderr))

info.append(f'stderr={_format_pipe(stderr)}')

logger.debug(' '.join(info))

async def subprocess_shell(self, protocol_factory, cmd, *,

popen_args = (program,) + args

for arg in popen_args:

if not isinstance(arg, (str, bytes)):

raise TypeError("program arguments must be "

"a bytes or text string, not %s"

% type(arg).__name__)

raise TypeError(

f"program arguments must be a bytes or text string, "

f"not {type(arg).__name__}")

protocol = protocol_factory()

if self._debug:

# don't log parameters: they may contain sensitive information

# (password) and may be too long

debug_log = 'execute program %r' % program

debug_log = f'execute program {program!r}'

self._log_subprocess(debug_log, stdin, stdout, stderr)

transport = await self._make_subprocess_transport(

protocol, popen_args, False, stdin, stdout, stderr,

documentation for details about context).

"""

if handler is not None and not callable(handler):

raise TypeError('A callable object or None is expected, '

'got {!r}'.format(handler))

raise TypeError(f'A callable object or None is expected, '

f'got {handler!r}')

self._exception_handler = handler

def default_exception_handler(self, context):

else:

exc_info = False

if ('source_traceback' not in context

and self._current_handle is not None

and self._current_handle._source_traceback):

context['handle_traceback'] = self._current_handle._source_traceback

if ('source_traceback' not in context and

self._current_handle is not None and

self._current_handle._source_traceback):

context['handle_traceback'] = \

self._current_handle._source_traceback

log_lines = [message]

for key in sorted(context):

value += tb.rstrip()

else:

value = repr(value)

log_lines.append('{}: {}'.format(key, value))

log_lines.append(f'{key}: {value}')

logger.error('\n'.join(log_lines), exc_info=exc_info)

if enabled:

if current_wrapper not in (None, wrapper):

warnings.warn(

"loop.set_debug(True): cannot set debug coroutine "

"wrapper; another wrapper is already set %r" %

current_wrapper, RuntimeWarning)

f"loop.set_debug(True): cannot set debug coroutine "

f"wrapper; another wrapper is already set "

f"{current_wrapper!r}",

RuntimeWarning)

else:

set_wrapper(wrapper)

self._coroutine_wrapper_set = True

else:

if current_wrapper not in (None, wrapper):

warnings.warn(

"loop.set_debug(False): cannot unset debug coroutine "

"wrapper; another wrapper was set %r" %

current_wrapper, RuntimeWarning)

f"loop.set_debug(False): cannot unset debug coroutine "

f"wrapper; another wrapper was set {current_wrapper!r}",

RuntimeWarning)

else:

set_wrapper(None)

self._coroutine_wrapper_set = False