Patch asyncio module to use pure Python tasks and futures.
(*, error_on_mispatched=False)
| 89 | |
| 90 | |
| 91 | def _patch_asyncio(*, error_on_mispatched=False): |
| 92 | """Patch asyncio module to use pure Python tasks and futures.""" |
| 93 | |
| 94 | def _get_event_loop(stacklevel=3): |
| 95 | loop = events._get_running_loop() |
| 96 | if loop is None: |
| 97 | if sys.version_info < (3, 14, 0): |
| 98 | policy = events.get_event_loop_policy() |
| 99 | else: |
| 100 | policy = events._get_event_loop_policy() |
| 101 | loop = policy.get_event_loop() |
| 102 | return loop |
| 103 | |
| 104 | if hasattr(asyncio, "_nest_patched"): |
| 105 | if not hasattr(asyncio, "_nest_asyncio"): |
| 106 | if error_on_mispatched: |
| 107 | raise RuntimeError("asyncio was already patched!") |
| 108 | elif sys.version_info >= (3, 12, 0): |
| 109 | import warnings |
| 110 | warnings.warn("asyncio was already patched!") |
| 111 | return |
| 112 | |
| 113 | asyncio.tasks.Task = asyncio.tasks._PyTask |
| 114 | asyncio.Task = asyncio.tasks._CTask = asyncio.tasks.Task |
| 115 | asyncio.Future = asyncio.futures._CFuture = asyncio.futures.Future = ( |
| 116 | asyncio.futures._PyFuture |
| 117 | ) |
| 118 | asyncio.get_event_loop = _get_event_loop |
| 119 | events._get_event_loop = events.get_event_loop = asyncio.get_event_loop |
| 120 | asyncio.run = run |
| 121 | asyncio._nest_patched = True |
| 122 | asyncio._nest_asyncio = PatchedNestAsyncio() |
| 123 | |
| 124 | |
| 125 | def _patch_policy(): |
no test coverage detected
searching dependent graphs…