()
| 24 | |
| 25 | @staticmethod |
| 26 | def _get_working_loop() -> asyncio.AbstractEventLoop: |
| 27 | try: |
| 28 | # Python <= 3.13: Returns existing or auto-creates a new loop. |
| 29 | # Python 3.14+: Returns existing or raises RuntimeError. |
| 30 | evloop = asyncio.get_event_loop() |
| 31 | except RuntimeError: |
| 32 | # No event loop exists in the current thread. |
| 33 | evloop = None |
| 34 | |
| 35 | gen_new_loop = not LoopValidator._is_valid_loop(evloop) |
| 36 | if gen_new_loop: |
| 37 | # Only attempt to close the loop if it actually exists |
| 38 | if evloop is not None and not evloop.is_closed(): |
| 39 | evloop.close() |
| 40 | |
| 41 | selector = selectors.SelectSelector() |
| 42 | new_loop = asyncio.SelectorEventLoop(selector) |
| 43 | asyncio.set_event_loop(new_loop) |
| 44 | return new_loop |
| 45 | |
| 46 | return evloop |
| 47 | |
| 48 | @staticmethod |
| 49 | def _is_valid_loop(evloop: asyncio.AbstractEventLoop) -> bool: |
no test coverage detected