INNER CODE UNIT · Python

task

ImFeH2/huddol · core/src/fora/adapters/websocket/server.py:243

    task = asyncio.create_task(asyncio.to_thread(operation, *args))
    try:
        return await asyncio.shield(task)
    except asyncio.CancelledError:
        await task
        raise


@asynccontextmanager
async def resource[T](
    operation: Callable[..., T],
    release: Callable[[T], None],
    *args: Any,
) -> AsyncIterator[T]:
    task = asyncio.create_task(asyncio.to_thread(operation, *args))
    try:
        result = await asyncio.shield(task)
        yield result

View source record →