DOCSEARCHbyAlgolia
Submit Docs

janus

repository·master·Indexed 21 days ago

https://github.com/aio-libs/janus

A thread-safe, mixed sync-async queue designed to bridge synchronous threaded code and asynchronous asyncio code. It provides a single queue object with two interfaces: a synchronous interface compatible with queue.Queue and an asynchronous interface following the asyncio.Queue design. Available queue types include janus.Queue (FIFO), janus.LifoQueue, and janus.PriorityQueue.

Tokens
858
Snippets
1
Records
3
Agent score
26%

What's inside janus

  1. What is Janus and when should I use it?

    master

    Janus is a mixed synchronous-asynchronous queue designed specifically for communication between classic synchronous (threaded) code and asynchronous (asyncio) code.

    It provides a single queue object with two distinct interfaces:

    1. A synchronous interface (sync_q) that is fully compatible with Python's standard queue.Queue.
    2. An asynchronous interface (async_q) that follows the asyncio.Queue design.

    Important Performance Note: Janus is optimized for cross-paradigm communication. If your application only uses synchronous code or only uses asynchronous code, you should use the standard queue or asyncio.Queue modules instead, as using Janus in those cases can result in significant performance slowdowns.

  2. Use Janus queues for sync-async communication

    master

    Janus provides three types of queues. Each queue instance exposes two properties: .sync_q for synchronous operations and .async_q for asynchronous operations.

    Available queue types:

    • janus.Queue (Standard FIFO)
    • janus.LifoQueue (Last-In, First-Out)
    • janus.PriorityQueue (Priority-based)

    To prevent asyncio error messages caused by background notification tasks, you must call .aclose() on the queue object once you are finished using it.

    import asyncio
import janus


def threaded(sync_q: janus.SyncQueue[int]) -> None:
    for i in range(100):
        sync_q.put(i)
    sync_q.join()


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


async def main() -> None:
    # Initialize the Janus queue
    queue: janus.Queue[int] = janus.Queue()
    
    loop = asyncio.get_running_loop()
    
    # Run the synchronous part in a thread executor
    fut = loop.run_in_executor(None, threaded, queue.sync_q)
    
    # Run the asynchronous part
    await async_coro(queue.async_q)
    
    await fut
    
    # CRITICAL: Always close the queue to clean up background tasks
    await queue.aclose()


asyncio.run(main())

  3. Janus limitations and constraints

    master

    When using Janus, be aware of the following constraints:

    • Lifecycle Management: You must call await queue.aclose() when finished. Failure to do so may result in asyncio error messages because the library creates internal tasks to notify threads.
    • Performance: Do not use Janus for sync-only or async-only workflows; use queue.Queue or asyncio.Queue respectively to avoid significant slowdowns.
    • Event Loop Binding: You cannot use a Janus queue to communicate between two different event loops. Like all asyncio primitives, the queue binds to the event loop that created it.