highlevel support for working with threads in asyncio import functools import contextvars from import events all tothread async def tothreadfunc args kwargs loop events getrunningloop ctx contextvars copycontext funccall functools partialctx run func args kwargs return await loop runinexecutornone funccall asynchronously run function func in a separate thread any args and kwargs supplied for this function are directly passed to func also the current class contextvars context is propagated allowing context variables from the main thread to be accessed in the separate thread return a coroutine that can be awaited to get the eventual result of func
import functools
import contextvars

from . import events


__all__ = "to_thread",


async def to_thread(func, /, *args, **kwargs):
    
    loop = events.get_running_loop()
    ctx = contextvars.copy_context()
    func_call = functools.partial(ctx.run, func, *args, **kwargs)
    return await loop.run_in_executor(None, func_call)