So, here's a prototype implementation of asyncio.ThreadPool that would function exactly as Yury described, but I'm not convinced about the design. Usually, it seems preferred to separate the context manager from the rest of the class (as was done with asyncio.Lock), but I think this one might be simple enough to be a single class:

class ThreadPool:
    def __init__(self, timeout=None):
        self.timeout = timeout
        self._loop = None
    
    async def __aenter__(self):
        self._loop = asyncio.get_running_loop()
        # Ensure that ThreadPoolExecutor is being used
        self._loop.default_executor = concurrent.futures.ThreadPoolExecutor()
        return self

    async def __aexit__(self, *args):
        await self._loop.shutdown_default_executor(timeout=self.timeout)

    def run(self, func, *args, **kwargs):
        call = functools.partial(func, *args, **kwargs)
        return self._loop.run_in_executor(None, call)

It utilizes the existing lower level event loop API to provide an asynchronous context manager. I'd say the main advantage is that it's significantly more user friendly, as there's no loop or executor argument to be specified, and users can freely pass kwargs to run() thanks to functools.partial().

Additionally, I also included a timeout parameter for shutting down the ThreadPoolExecutor, which will depend upon GH-16360. This can be used as such:

async def main():
    async with asyncio.ThreadPool(timeout=600) as pool:
        fut1 = pool.run(func)
        fut2 = pool.run(func, arg1, arg2)
        fut2 = pool.run(func, arg1, kwarg1=True)
    print(await asyncio.gather(fut1, fut2, fut3))

asyncio.run(main())

I don't expect that this would be the final design, but I think it's a decent prototype to demonstrate the functionality. Thoughts?