$ show aio-libs/janus
janus is a Python library that provides mixed sync-async queues for communication between threaded code and asyncio code. The core architecture centers on a single `Queue` class that internally manages both synchronous and asynchronous interfaces via proxy objects (`_SyncQueueProxy` and `_AsyncQueueProxy`). It uses thread-safe conditions (`threading.Condition`) for sync operations and asyncio conditions (`asyncio.Condition`) with careful cross-thread notification via `loop.call_soon_threadsafe`. The queue supports standard FIFO, LIFO (`LifoQueue`), and priority (`PriorityQueue`) variants through a template method pattern (overriding `_init`, `_put`, `_get`). Lifecycle management includes graceful and immediate shutdown with dedicated exception types (`SyncQueueShutDown`/`AsyncQueueShutDown`) compatible with Python 3.13+. The library is production-stable, has no external dependencies beyond the standard library, and is thoroughly tested across multiple Python versions and concurrency scenarios. It does not expose APIs beyond queue operations and is strictly a communication primitive.
The LLM identified 1 files worth highlighting for reuse:
The `SyncQueue` and `AsyncQueue` Protocol classes define abstract interfaces for queue operations, providing type-safe contracts that can be implemented by any queue implementation. They include `put`, `get`, `join`, `task_done`, `qsize`, `shutdown`, and property `closed`. These protocols are fully self-contained, import only standard library types, and can be extracted into a separate project to enable dependency injection or to define queue abstractions in other concurrency libraries. They demonstrate disciplined use of Python's `typing.Protocol` for structural subtyping, which is a non-obvious skill for designing testable and substitutable concurrency components.