1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
|
# Copyright (C) 2023 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
from PySide6.QtCore import (QCoreApplication, QDateTime, QDeadlineTimer,
QEventLoop, QObject, QTimer, QThread, Slot)
from . import futures
from . import tasks
import asyncio
import collections.abc
import concurrent.futures
import contextvars
import enum
import os
import signal
import socket
import subprocess
import typing
import warnings
__all__ = [
"QAsyncioEventLoopPolicy", "QAsyncioEventLoop",
"QAsyncioHandle", "QAsyncioTimerHandle",
]
class QAsyncioExecutorWrapper(QObject):
def __init__(self, func: typing.Callable, *args: typing.Tuple) -> None:
super().__init__()
self._loop: QEventLoop
self._func = func
self._args = args
self._result = None
self._exception = None
def _cb(self):
try:
self._result = self._func(*self._args)
except BaseException as e:
self._exception = e
self._loop.exit()
def do(self):
# This creates a new event loop and dispatcher for the thread, if not already created.
self._loop = QEventLoop()
asyncio.events._set_running_loop(self._loop)
QTimer.singleShot(0, self._loop, lambda: self._cb())
self._loop.exec()
if self._exception is not None:
raise self._exception
return self._result
def exit(self):
self._loop.exit()
class QAsyncioEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
def __init__(self,
application: typing.Optional[QCoreApplication] = None,
quit_qapp: bool = True) -> None:
super().__init__()
if application is None:
if QCoreApplication.instance() is None:
application = QCoreApplication()
else:
application = QCoreApplication.instance()
self._application: QCoreApplication = application # type: ignore[assignment]
self._quit_qapp = quit_qapp
self._event_loop: typing.Optional[asyncio.AbstractEventLoop] = None
signal.signal(signal.SIGINT, signal.SIG_DFL)
def get_event_loop(self) -> asyncio.AbstractEventLoop:
if self._event_loop is None:
self._event_loop = QAsyncioEventLoop(self._application)
return self._event_loop
def set_event_loop(self, loop: typing.Optional[asyncio.AbstractEventLoop]) -> None:
self._event_loop = loop
def new_event_loop(self) -> asyncio.AbstractEventLoop:
return QAsyncioEventLoop(self._application, quit_qapp=self._quit_qapp)
def get_child_watcher(self) -> asyncio.AbstractChildWatcher:
raise DeprecationWarning("Child watchers are deprecated since Python 3.12")
def set_child_watcher(self, watcher: asyncio.AbstractChildWatcher) -> None:
raise DeprecationWarning("Child watchers are deprecated since Python 3.12")
class QAsyncioEventLoop(asyncio.BaseEventLoop, QObject):
"""
Implements the asyncio API:
https://docs.python.org/3/library/asyncio-eventloop.html
"""
class ShutDownThread(QThread):
def __init__(self, future: futures.QAsyncioFuture, loop: "QAsyncioEventLoop") -> None:
super().__init__()
self._future = future
self._loop = loop
self.started.connect(self.shutdown)
def run(self) -> None:
pass
def shutdown(self) -> None:
try:
self._loop._default_executor.shutdown(wait=True)
if not self._loop.is_closed():
self._loop.call_soon_threadsafe(self._future.set_result, None)
except Exception as e:
if not self._loop.is_closed():
self._loop.call_soon_threadsafe(self._future.set_exception, e)
def __init__(self,
application: QCoreApplication, quit_qapp: bool = True) -> None:
asyncio.BaseEventLoop.__init__(self)
QObject.__init__(self)
self._application: QCoreApplication = application
self._quit_qapp = quit_qapp
self._thread = QThread.currentThread()
self._closed = False
self._quit_from_inside = False
self._quit_from_outside = False
self._asyncgens: typing.Set[collections.abc.AsyncGenerator] = set()
# Starting with Python 3.11, this must be an instance of
# ThreadPoolExecutor.
self._default_executor = concurrent.futures.ThreadPoolExecutor()
self._exception_handler: typing.Optional[typing.Callable] = self.default_exception_handler
self._task_factory: typing.Optional[typing.Callable] = None
self._future_to_complete: typing.Optional[futures.QAsyncioFuture] = None
self._debug = bool(os.getenv("PYTHONASYNCIODEBUG", False))
self._application.aboutToQuit.connect(self._about_to_quit_cb)
# Running and stopping the loop
def _run_until_complete_cb(self, future: futures.QAsyncioFuture) -> None:
if not future.cancelled():
if isinstance(future.exception(), (SystemExit, KeyboardInterrupt)):
return
future.get_loop().stop()
def run_until_complete(self,
future: futures.QAsyncioFuture) -> typing.Any: # type: ignore[override]
if self.is_closed():
raise RuntimeError("Event loop is closed")
if self.is_running():
raise RuntimeError("Event loop is already running")
arg_was_coro = not asyncio.futures.isfuture(future)
future = asyncio.tasks.ensure_future(future, loop=self) # type: ignore[assignment]
future.add_done_callback(self._run_until_complete_cb)
self._future_to_complete = future
try:
self.run_forever()
except Exception as e:
if arg_was_coro and future.done() and not future.cancelled():
future.exception()
raise e
finally:
future.remove_done_callback(self._run_until_complete_cb)
if not future.done():
raise RuntimeError("Event loop stopped before Future completed")
return future.result()
def run_forever(self) -> None:
if self.is_closed():
raise RuntimeError("Event loop is closed")
if self.is_running():
raise RuntimeError("Event loop is already running")
asyncio.events._set_running_loop(self)
self._application.exec()
asyncio.events._set_running_loop(None)
def _about_to_quit_cb(self):
if not self._quit_from_inside:
self._quit_from_outside = True
def stop(self) -> None:
if self._future_to_complete is not None:
if self._future_to_complete.done():
self._future_to_complete = None
else:
return
self._quit_from_inside = True
if self._quit_qapp:
self._application.quit()
def is_running(self) -> bool:
return self._thread.loopLevel() > 0
def is_closed(self) -> bool:
return self._closed
def close(self) -> None:
if self.is_running():
raise RuntimeError("Cannot close a running event loop")
if self.is_closed():
return
if self._default_executor is not None:
self._default_executor.shutdown(wait=False)
if self._quit_qapp:
self._application.shutdown()
self._closed = True
async def shutdown_asyncgens(self) -> None:
if not len(self._asyncgens):
return
results = await asyncio.tasks.gather(
*[asyncgen.aclose() for asyncgen in self._asyncgens],
return_exceptions=True)
for result, asyncgen in zip(results, self._asyncgens):
if isinstance(result, Exception):
self.call_exception_handler({
"message": f"Closing asynchronous generator {asyncgen}"
f"raised an exception",
"exception": result,
"asyncgen": asyncgen})
self._asyncgens.clear()
async def shutdown_default_executor(self, # type: ignore[override]
timeout: typing.Union[int, float, None] = None) -> None:
shutdown_successful = False
if timeout is not None:
deadline_timer = QDeadlineTimer(int(timeout * 1000))
else:
deadline_timer = QDeadlineTimer(QDeadlineTimer.Forever)
if self._default_executor is None:
return
future = self.create_future()
thread = QAsyncioEventLoop.ShutDownThread(future, self)
thread.start()
try:
await future
finally:
shutdown_successful = thread.wait(deadline_timer)
if timeout is not None and not shutdown_successful:
warnings.warn(
f"Could not shutdown the default executor within {timeout} seconds",
RuntimeWarning, stacklevel=2)
self._default_executor.shutdown(wait=False)
# Scheduling callbacks
def _call_soon_impl(self, callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None,
is_threadsafe: typing.Optional[bool] = False) -> asyncio.Handle:
return self._call_later_impl(0, callback, *args, context=context,
is_threadsafe=is_threadsafe)
def call_soon(self, callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None) -> asyncio.Handle:
return self._call_soon_impl(callback, *args, context=context, is_threadsafe=False)
def call_soon_threadsafe(self, callback: typing.Callable, *args: typing.Any,
context:
typing.Optional[contextvars.Context] = None) -> asyncio.Handle:
if context is None:
context = contextvars.copy_context()
return self._call_soon_impl(callback, *args, context=context, is_threadsafe=True)
def _call_later_impl(self, delay: typing.Union[int, float],
callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None,
is_threadsafe: typing.Optional[bool] = False) -> asyncio.TimerHandle:
if not isinstance(delay, (int, float)):
raise TypeError("delay must be an int or float")
return self._call_at_impl(self.time() + delay, callback, *args, context=context,
is_threadsafe=is_threadsafe)
def call_later(self, delay: typing.Union[int, float],
callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None) -> asyncio.TimerHandle:
return self._call_later_impl(delay, callback, *args, context=context, is_threadsafe=False)
def _call_at_impl(self, when: typing.Union[int, float],
callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None,
is_threadsafe: typing.Optional[bool] = False) -> asyncio.TimerHandle:
if not isinstance(when, (int, float)):
raise TypeError("when must be an int or float")
if self.is_closed():
raise RuntimeError("Event loop is closed")
return QAsyncioTimerHandle(when, callback, args, self, context, is_threadsafe=is_threadsafe)
def call_at(self, when: typing.Union[int, float],
callback: typing.Callable, *args: typing.Any,
context: typing.Optional[contextvars.Context] = None) -> asyncio.TimerHandle:
return self._call_at_impl(when, callback, *args, context=context, is_threadsafe=False)
def time(self) -> float:
return QDateTime.currentMSecsSinceEpoch() / 1000
# Creating Futures and Tasks
def create_future(self) -> futures.QAsyncioFuture: # type: ignore[override]
return futures.QAsyncioFuture(loop=self)
def create_task(self, # type: ignore[override]
coro: typing.Union[collections.abc.Generator, collections.abc.Coroutine],
*, name: typing.Optional[str] = None,
context: typing.Optional[contextvars.Context] = None) -> tasks.QAsyncioTask:
if self.is_closed():
raise RuntimeError("Event loop is closed")
if self._task_factory is None:
task = tasks.QAsyncioTask(coro, loop=self, name=name, context=context)
else:
task = self._task_factory(self, coro, context=context)
task.set_name(name)
return task
def set_task_factory(self, factory: typing.Optional[typing.Callable]) -> None:
if factory is not None and not callable(factory):
raise TypeError("The task factory must be a callable or None")
self._task_factory = factory
def get_task_factory(self) -> typing.Optional[typing.Callable]:
return self._task_factory
# Opening network connections
async def create_connection(
self, protocol_factory, host=None, port=None,
*, ssl=None, family=0, proto=0,
flags=0, sock=None, local_addr=None,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
happy_eyeballs_delay=None, interleave=None):
raise NotImplementedError
async def create_datagram_endpoint(self, protocol_factory,
local_addr=None, remote_addr=None, *,
family=0, proto=0, flags=0,
reuse_address=None, reuse_port=None,
allow_broadcast=None, sock=None):
raise NotImplementedError
async def create_unix_connection(
self, protocol_factory, path=None, *,
ssl=None, sock=None,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
raise NotImplementedError
# Creating network servers
async def create_server(
self, protocol_factory, host=None, port=None,
*, family=socket.AF_UNSPEC,
flags=socket.AI_PASSIVE, sock=None, backlog=100,
ssl=None, reuse_address=None, reuse_port=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
raise NotImplementedError
async def create_unix_server(
self, protocol_factory, path=None, *,
sock=None, backlog=100, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
raise NotImplementedError
async def connect_accepted_socket(
self, protocol_factory, sock,
*, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
raise NotImplementedError
# Transferring files
async def sendfile(self, transport, file, offset=0, count=None,
*, fallback=True):
raise NotImplementedError
# TLS Upgrade
async def start_tls(self, transport, protocol, sslcontext, *,
server_side=False,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
raise NotImplementedError
# Watching file descriptors
def add_reader(self, fd, callback, *args):
raise NotImplementedError
def remove_reader(self, fd):
raise NotImplementedError
def add_writer(self, fd, callback, *args):
raise NotImplementedError
def remove_writer(self, fd):
raise NotImplementedError
# Working with socket objects directly
async def sock_recv(self, sock, nbytes):
raise NotImplementedError
async def sock_recv_into(self, sock, buf):
raise NotImplementedError
async def sock_recvfrom(self, sock, bufsize):
raise NotImplementedError
async def sock_recvfrom_into(self, sock, buf, nbytes=0):
raise NotImplementedError
async def sock_sendall(self, sock, data):
raise NotImplementedError
async def sock_sendto(self, sock, data, address):
raise NotImplementedError
async def sock_connect(self, sock, address):
raise NotImplementedError
async def sock_accept(self, sock):
raise NotImplementedError
async def sock_sendfile(self, sock, file, offset=0, count=None, *,
fallback=None):
raise NotImplementedError
# DNS
async def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0):
raise NotImplementedError
async def getnameinfo(self, sockaddr, flags=0):
raise NotImplementedError
# Working with pipes
async def connect_read_pipe(self, protocol_factory, pipe):
raise NotImplementedError
async def connect_write_pipe(self, protocol_factory, pipe):
raise NotImplementedError
# Unix signals
def add_signal_handler(self, sig, callback, *args):
raise NotImplementedError
def remove_signal_handler(self, sig):
raise NotImplementedError
# Executing code in thread or process pools
def run_in_executor(self,
executor: typing.Optional[concurrent.futures.ThreadPoolExecutor],
func: typing.Callable, *args: typing.Tuple) -> asyncio.futures.Future:
if self.is_closed():
raise RuntimeError("Event loop is closed")
if executor is None:
executor = self._default_executor
wrapper = QAsyncioExecutorWrapper(func, *args)
return asyncio.futures.wrap_future(
executor.submit(wrapper.do), loop=self
)
def set_default_executor(self,
executor: typing.Optional[
concurrent.futures.ThreadPoolExecutor]) -> None:
if not isinstance(executor, concurrent.futures.ThreadPoolExecutor):
raise TypeError("The executor must be a ThreadPoolExecutor")
self._default_executor = executor
# Error Handling API
def set_exception_handler(self, handler: typing.Optional[typing.Callable]) -> None:
if handler is not None and not callable(handler):
raise TypeError("The handler must be a callable or None")
self._exception_handler = handler
def get_exception_handler(self) -> typing.Optional[typing.Callable]:
return self._exception_handler
def default_exception_handler(self, context: typing.Dict[str, typing.Any]) -> None:
# TODO
if context["message"]:
print(context["message"])
def call_exception_handler(self, context: typing.Dict[str, typing.Any]) -> None:
if self._exception_handler is not None:
self._exception_handler(context)
# Enabling debug mode
def get_debug(self) -> bool:
# TODO: Part of the asyncio API but currently unused. More details:
# https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode
return self._debug
def set_debug(self, enabled: bool) -> None:
self._debug = enabled
# Running subprocesses
async def subprocess_exec(self, protocol_factory, *args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
**kwargs):
raise NotImplementedError
async def subprocess_shell(self, protocol_factory, cmd, *,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
**kwargs):
raise NotImplementedError
class QAsyncioHandle():
class HandleState(enum.Enum):
PENDING = enum.auto()
CANCELLED = enum.auto()
DONE = enum.auto()
def __init__(self, callback: typing.Callable, args: typing.Tuple,
loop: QAsyncioEventLoop, context: typing.Optional[contextvars.Context],
is_threadsafe: typing.Optional[bool] = False) -> None:
self._callback = callback
self._args = args
self._loop = loop
self._context = context
self._is_threadsafe = is_threadsafe
self._timeout = 0
self._state = QAsyncioHandle.HandleState.PENDING
self._start()
def _schedule_event(self, timeout: int, func: typing.Callable) -> None:
if not self._loop.is_closed() and not self._loop._quit_from_outside:
if self._is_threadsafe:
QTimer.singleShot(timeout, self._loop, func)
else:
QTimer.singleShot(timeout, func)
def _start(self) -> None:
self._schedule_event(self._timeout, lambda: self._cb())
@Slot()
def _cb(self) -> None:
if self._state == QAsyncioHandle.HandleState.PENDING:
if self._context is not None:
self._context.run(self._callback, *self._args)
else:
self._callback(*self._args)
self._state = QAsyncioHandle.HandleState.DONE
def cancel(self) -> None:
if self._state == QAsyncioHandle.HandleState.PENDING:
# The old timer that was created in _start will still trigger but _cb won't do anything.
self._state = QAsyncioHandle.HandleState.CANCELLED
def cancelled(self) -> bool:
return self._state == QAsyncioHandle.HandleState.CANCELLED
class QAsyncioTimerHandle(QAsyncioHandle, asyncio.TimerHandle):
def __init__(self, when: float, callback: typing.Callable, args: typing.Tuple,
loop: QAsyncioEventLoop, context: typing.Optional[contextvars.Context],
is_threadsafe: typing.Optional[bool] = False) -> None:
QAsyncioHandle.__init__(self, callback, args, loop, context, is_threadsafe)
self._when = when
self._timeout = int(max(self._when - self._loop.time(), 0) * 1000)
QAsyncioHandle._start(self)
# Override this so that timer.start() is only called once at the end
# of the constructor for both QtHandle and QtTimerHandle.
def _start(self) -> None:
pass
def when(self) -> float:
return self._when
|