Run a synchronous function in a thread pool to avoid blocking the event loop.
(func: Callable[..., _T], *args)
| 9 | |
| 10 | |
| 11 | async def run_sync_in_thread(func: Callable[..., _T], *args) -> _T: |
| 12 | """Run a synchronous function in a thread pool to avoid blocking the event loop.""" |
| 13 | return await asyncio.to_thread(func, *args) |
| 14 | |
| 15 | |
| 16 | if TYPE_CHECKING: |
no outgoing calls
no test coverage detected