What is the correct way to type annotate a janus.Queue?

[rbuffat]

When I try to add type hints to the example, pylance creates reportPrivateUsage warnings.

import asyncio
import janus


def threaded(sync_q: janus._SyncQueueProxy[int]):
    for i in range(100):
        sync_q.put(i)
    sync_q.join()


async def async_coro(async_q: janus._AsyncQueueProxy[int]):
    for i in range(100):
        val = await async_q.get()
        assert val == i
        async_q.task_done()


async def main():
    queue: janus.Queue[int] = janus.Queue()
    loop = asyncio.get_running_loop()
    fut = loop.run_in_executor(None, threaded, queue.sync_q)
    await async_coro(queue.async_q)
    await fut
    queue.close()
    await queue.wait_closed()


asyncio.run(main())

Warnings:

"_SyncQueueProxy" is private and used outside of the module in which it is declared

"_AsyncQueueProxy" is private and used outside of the module in which it is declared

What is the preferred way to add type hints to janus Queues?

[asvetlov]

Good question!

Perhaphs we should make both classes public but forbid their instantiation by user code.

I can imagine something like NoPublicConstructor approach

Would you make a pull request?

[rbuffat]

I will give it a shot, but I probably need some time to get familiar with it.

[asvetlov]

Fixed by #374