Trio support in the future?

[jupiterbjy]

Trio, a structured concurrency asynchronous library - is now gaining more and more library supports.

And I love using loguru, but sadly it's async support is asyncio flavored and doesn't work with trio.

from sys import version
import asyncio

from loguru import logger
import trio


print(version)


async def asyncio_sink(message):
    print(f"Asyncio sink start")
    await asyncio.sleep(2)
    print(f"Asyncio Delayed message - {message}")


async def trio_sink(message):
    print(f"Trio sink start")
    await trio.sleep(2)
    print(f"Trio Delayed message - {message}")


async def asyncio_test():
    logger.remove()
    logger.add(asyncio_sink)

    logger.debug("Meow")
    await logger.complete()


async def trio_test():
    logger.remove()
    logger.add(trio_sink)

    logger.debug("Meow")
    await logger.complete()


asyncio.run(asyncio_test())
trio.run(trio_test)


# Output
"""
3.10.2 (tags/v3.10.2:a58ebcc, Jan 17 2022, 14:12:15) [MSC v.1929 64 bit (AMD64)]
Asyncio sink start
Asyncio Delayed message - 2022-02-02 19:44:01.149 | DEBUG    | __main__:asyncio_test:27 - Meow

--- Logging error in Loguru Handler #2 ---
Record was: {'elapsed': datetime.timedelta(seconds=2, microseconds=345309), 'exception': None, 'extra': {}, 'file': (name='scratch.py', path='C:\\Users\\Nyarukoishi\\AppData\\Roaming\\JetBrains\\PyCharm2021.3\\scratches\\scratch.py'), 'function': 'trio_test', 'level': (name='DEBUG', no=10, icon='🐞'), 'line': 35, 'message': 'Meow', 'module': 'scratch', 'name': '__main__', 'process': (id=19688, name='MainProcess'), 'thread': (id=11420, name='MainThread'), 'time': datetime(2022, 2, 2, 19, 44, 3, 223969, tzinfo=datetime.timezone(datetime.timedelta(seconds=32400), '대한민국 표준시'))}
Traceback (most recent call last):
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_handler.py", line 177, in emit
    self._sink.write(str_record)
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_simple_sinks.py", line 78, in write
    loop = self._loop or asyncio.get_event_loop()
  File "C:\python310\lib\asyncio\events.py", line 656, in get_event_loop
    raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'MainThread'.
--- End of logging error ---
D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_handler.py:182: RuntimeWarning: coroutine 'trio_sink' was never awaited
  self._error_interceptor.print(record)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
Traceback (most recent call last):
  File "C:\Users\Nyarukoishi\AppData\Roaming\JetBrains\PyCharm2021.3\scratches\scratch.py", line 40, in <module>
    trio.run(trio_test)
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\trio\_core\_run.py", line 1932, in run
    raise runner.main_task_outcome.error
  File "C:\Users\Nyarukoishi\AppData\Roaming\JetBrains\PyCharm2021.3\scratches\scratch.py", line 36, in trio_test
    await logger.complete()
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_logger.py", line 1073, in __await__
    yield from handler.complete_async().__await__()
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_handler.py", line 209, in complete_async
    await self._sink.complete()
  File "D:\github\StreamNotifier-Web\venv\lib\site-packages\loguru\_simple_sinks.py", line 96, in complete
    loop = asyncio.get_event_loop()
  File "C:\python310\lib\asyncio\events.py", line 656, in get_event_loop
    raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'MainThread'.

Process finished with exit code 1
"""

So in TL;DR - Will loguru support trio eventually?

Ideally supporting asyncio, trio, anyio, curio all in one like HTTPX would be the absolute best, but after looking their source codes I don't think that's an easy work.

---

Workaround

I'm leaving here my workaround for anyone who want to use loguru with trio without relying on translation layer like trio-asyncio

from loguru import logger
import trio


send_ch, recv_ch = trio.open_memory_channel(256)


def trio_sink(message):
    try:
        send_ch.send_nowait(message)
    except trio.WouldBlock:
        pass  # queue(memory channel) is full, can't put more)


async def trio_sink_async(message):
    print(f"Trio sink start")
    await trio.sleep(2)
    print(f"Trio Delayed message - {message}")


async def trio_sink_task(
    mem_receive_channel: trio.MemoryReceiveChannel, task_status=trio.TASK_STATUS_IGNORED
):
    task_status.started()

    async for message in mem_receive_channel:
        await trio_sink_async(message)


async def your_main_code_here():
    logger.debug("Meow")


async def boilerplate():
    logger.remove()
    logger.add(trio_sink)

    async with trio.open_nursery() as nursery:
        await nursery.start(trio_sink_task, recv_ch)

        try:
            await your_main_code_here()
        finally:
            await send_ch.aclose()


trio.run(boilerplate)

# Output
"""
Trio sink start
Trio Delayed message - 2022-02-02 22:23:50.149 | DEBUG    | __main__:your_main_code_here:31 - Meow

Process finished with exit code 0
"""

[Delgan]

Hi @jupiterbjy, thanks for bringing this up. I knew trio but never tried it with Loguru and didn't now both libraries weren't compatible. Trio is a quite popular alternative to asyncio and I would love it to work with Loguru. I have no idea how to do this for now, but I'll look into it.