janus

library2.0.0pypypi✓ verified 29d ago

janus is a Python library providing mixed synchronous and asynchronous queues to facilitate communication between classic threaded code and asyncio tasks. It offers `Queue`, `LifoQueue`, and `PriorityQueue` implementations, each with distinct synchronous (`.sync_q`) and asynchronous (`.async_q`) interfaces. The current version is 2.0.0, and the library is actively maintained to support the latest Python versions.

http-networking web-framework ai-ml llm-agents database

pip install janus

github ↗package ↗

INSTALL

IMPORT

SIG · JANUS

janus

http-networkingpythonv2.0.0

Install

1.6s avg

Import

203ms

Disk

16MB

Pass rate

10/ 10

Env Coverage10 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v2.0.0 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

py 3.10–3.95 runs

installs and imports cleanly · install 0.0s · import 0.208s · 17.9MB

glibc

py 3.10–3.95 runs

installs and imports cleanly · install 1.6s · import 0.198s · 18MB

16MB installed

● package 16MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

Queue

✓ from janus import Queue

LifoQueue

✓ from janus import LifoQueue

PriorityQueue

✓ from janus import PriorityQueue

SyncQueue

✓ from janus import SyncQueue

AsyncQueue

✓ from janus import AsyncQueue

SyncQueueEmpty

✓ from janus import SyncQueueEmpty

✗ from queue import Empty; # For janus.SyncQueue

As of v1.1.0, janus re-exports its own specific QueueEmpty/Full exceptions, which should be used for janus queues.

AsyncQueueEmpty

✓ from janus import AsyncQueueEmpty

✗ from asyncio.queues import QueueEmpty; # For janus.AsyncQueue

As of v1.1.0, janus re-exports its own specific QueueEmpty/Full exceptions, which should be used for janus queues.

This example demonstrates how to use `janus.Queue` to allow a synchronous thread to put items into a queue while an asynchronous coroutine concurrently consumes them. It highlights the use of `.sync_q` and `.async_q` interfaces and the important `aclose()` call for cleanup.

import asyncio import threading import janus def threaded_producer(sync_q: janus.SyncQueue[int]) -> None: print("Thread: Starting producer") for i in range(5): print(f"Thread: Putting {i}") sync_q.put(i) sync_q.join() # Wait for all items to be processed by async_coro print("Thread: Producer finished and joined") async def async_consumer(async_q: janus.AsyncQueue[int]) -> None: print("Async: Starting consumer") for _ in range(5): val = await async_q.get() print(f"Async: Got {val}") async_q.task_done() print("Async: Consumer finished") async def main() -> None: queue: janus.Queue[int] = janus.Queue() loop = asyncio.get_running_loop() # Run the synchronous producer in a separate thread producer_thread = threading.Thread(target=threaded_producer, args=(queue.sync_q,)) producer_thread.start() # Run the asynchronous consumer await async_consumer(queue.async_q) producer_thread.join() # Ensure thread completes before closing queue await queue.aclose() # Crucial for proper shutdown of janus resources print("Main: Queue closed") if __name__ == '__main__': asyncio.run(main())

Debug

Known issues

breakingIn v2.0.0, the `shutdown()` method's error handling changed. Calling `shutdown()` on a closed queue now raises `janus.AsyncQueueShutDown` or `janus.SyncQueueShutDown` instead of `RuntimeError`. Additionally, `task_done()` and `join()` methods no longer raise exceptions on queue shutdown/closing, aligning with stdlib queue behavior. This may require updating exception handling logic. [cite: Release Notes]

fix

Update `try...except RuntimeError` blocks to catch `janus.AsyncQueueShutDown` or `janus.SyncQueueShutDown` when dealing with queue shutdowns. Adjust code that previously relied on `task_done()` or `join()` raising exceptions after a queue is closed.

affects: >=2.0.0

breakingVersion 1.1.0 dropped support for Python 3.7 and 3.8. The library now requires Python 3.9 or higher. [cite: Release Notes]

fix

Upgrade your Python environment to 3.9 or a newer supported version.

affects: >=1.1.0

breakingAs of v0.5.0, explicit `loop` arguments were removed from `janus.Queue()` instantiation, and it became forbidden to create queues outside an active asyncio event loop. The library now automatically uses `asyncio.get_running_loop()`. [cite: Release Notes]

fix

Remove the `loop` argument when creating `janus.Queue` instances (e.g., `janus.Queue(loop=my_loop)` should become `janus.Queue()`). Ensure queue creation happens within an `asyncio` context where `asyncio.get_running_loop()` returns a valid loop.

affects: >=0.5.0

gotchaIt is crucial to call `await queue.aclose()` when you are finished with a `janus.Queue`. Failure to do so can lead to `asyncio` generating error messages or resource leaks, as the library creates internal tasks to manage notifications.

fix

Always include `await queue.aclose()` in your cleanup logic, typically at the end of your `async` entry point, to properly shut down internal resources.

affects: All

gotchajanus queues are specifically designed for interoperation between synchronous threads and asynchronous asyncio tasks. For purely synchronous (thread-to-thread) or purely asynchronous (asyncio-to-asyncio) communication, using standard `queue.Queue` or `asyncio.Queue` respectively is recommended, as `janus` can introduce significant slowdowns in these single-mode scenarios.

fix

Evaluate your use case: if communication is strictly sync-to-sync or async-to-async, prefer Python's built-in `queue` or `asyncio.Queue` modules for better performance.

affects: All

gotchajanus queues cannot be used for communication between two *different* asyncio event loops. Like other asyncio primitives, they are bound to the specific event loop in which they are created.

fix

Ensure that `janus` queues are created and accessed within the context of a single asyncio event loop. For inter-process or inter-event loop communication, consider higher-level mechanisms like multiprocessing queues or message brokers.

affects: All

Errors

Common errors & fixes

ModuleNotFoundError: No module named 'janus'

The 'janus' library has not been installed in your Python environment.

fix

Run `pip install janus` in your terminal to install the library.

RuntimeError: bound to a different event loop

A `janus.Queue` instance is tightly bound to the `asyncio` event loop in which it was created and cannot be used with a different event loop in another thread.

fix

Ensure that each `janus.Queue` is created and used exclusively within the same `asyncio` event loop, typically by creating it in the thread where its asynchronous interface will be primarily accessed.

AttributeError: 'Queue' object has no attribute 'put'

You are attempting to call `put()` or `get()` directly on a `janus.Queue` object, but these methods are only available through its synchronous (`.sync_q`) or asynchronous (`.async_q`) interfaces.

fix

Access the `put()` or `get()` methods via the appropriate interface: `queue.sync_q.put(...)` for synchronous operations or `await queue.async_q.put(...)` for asynchronous operations.

TypeError: 'function' object is not awaitable

You are attempting to `await` a synchronous method, specifically calling `await` on `put()` or `get()` from the synchronous interface (`.sync_q`) of a `janus.Queue`.

fix

Remove `await` when calling methods on the synchronous interface (e.g., `queue.sync_q.put(item)`), or use the asynchronous interface with `await` (e.g., `await queue.async_q.put(item)`).

Upgrade

Version history

2.0.0latest on PyPI · released Dec 13, 2024

Audit

Dependencies

No dependency data recorded yet.

Agent activity

28 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/aio-libs/janus ↗

packagehttps://pypi.org/project/janus/ ↗