(main, *, debug=False)
| 38 | |
| 39 | if sys.version_info < (3, 12, 0): |
| 40 | def run(main, *, debug=False): |
| 41 | loop = asyncio.get_event_loop() |
| 42 | loop.set_debug(debug) |
| 43 | task = asyncio.ensure_future(main) |
| 44 | try: |
| 45 | return loop.run_until_complete(task) |
| 46 | finally: |
| 47 | if not task.done(): |
| 48 | task.cancel() |
| 49 | with suppress(asyncio.CancelledError): |
| 50 | loop.run_until_complete(task) |
| 51 | else: |
| 52 | def run(main, *, debug=False, loop_factory=None): |
| 53 | new_event_loop = False |
nothing calls this directly
no test coverage detected