aiochannel

library1.3.0pypypi✓ verified 85d ago

aiochannel provides asyncio-compatible channels (closable queues) inspired by Go's concurrency model. It extends the functionality of `asyncio.Queue` by introducing a concept of a channel being 'closed and drained', guaranteeing that no further items can be added once closed. The library is active, with the current version being 1.3.0, and typically sees releases as needed for fixes or minor enhancements.

web-framework http-networking

pip install aiochannel

github ↗package ↗

INSTALL

IMPORT

SIG · AIOCHANNEL

aiochannel

web-frameworkpythonv1.3.0

Install

1.6s avg

Import

266ms

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 v1.3.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.920 runs

installs and imports cleanly · install 0.0s · import 0.284s · 17.8MB

glibc

py 3.10–3.920 runs

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

16MB installed

● package 16MB

Code

Verified usage

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

Channel

✓ from aiochannel import Channel

This quickstart demonstrates creating an `aiochannel.Channel`, with a producer coroutine putting items and a consumer coroutine retrieving them using an `async for` loop. The producer closes the channel, and the `channel.join()` call ensures the main program waits until all items are consumed and the channel is fully drained.

import asyncio from aiochannel import Channel, ChannelClosed async def producer(ch: Channel, num_items: int): for i in range(num_items): await ch.put(f"item-{i}") print(f"Produced item-{i}") ch.close() print("Producer closed the channel") async def consumer(ch: Channel): try: async for item in ch: print(f"Consumed {item}") except ChannelClosed: print("Consumer detected channel closed and drained.") async def main(): channel = Channel(10) # Create a channel with a buffer size of 10 await asyncio.gather( producer(channel, 5), consumer(channel) ) # .join() waits until the channel is both closed and drained await channel.join() print("Channel is closed and drained, main exiting.") if __name__ == "__main__": asyncio.run(main())

Debug

Known issues

gotchaOnce an `aiochannel.Channel` is closed using `channel.close()`, it is a permanent state and cannot be reopened. Subsequent calls to `put()` will raise `ChannelClosed`.

fix

Design your application logic so that channel closure is an irreversible signal for no more data. If a channel needs to be 'reset', create a new `Channel` instance.

affects: All versions

gotchaThe `channel.join()` method in `aiochannel.Channel` behaves differently from `asyncio.Queue.join()`. `aiochannel.Channel.join()` waits until the channel is *both* closed (`.close()` has been called) and completely drained (all items have been retrieved). `asyncio.Queue.join()` only waits until the queue is empty.

fix

Always use `channel.join()` when you need to ensure all processing is complete and the channel is empty, after calling `channel.close()`. Do not rely on `channel.empty()` alone for this purpose if `close()` is involved.

affects: All versions

gotcha`aiochannel.Channel` does not implement `task_done()` or rely on it for its `join()` semantics, unlike `asyncio.Queue`. The completion signal for `aiochannel` is its `closed and drained` state.

fix

Do not attempt to call `task_done()` on an `aiochannel.Channel` instance as it is not part of its API. Rely on `channel.close()` and `channel.join()` for managing completion.

affects: All versions

deprecatedExplicitly passing an `asyncio` event loop instance (e.g., `Channel(loop=my_loop)`) to the `Channel` constructor is generally discouraged as `asyncio` itself is phasing out explicit loop passing in favor of `asyncio.get_running_loop()`.

fix

Omit the `loop` argument when creating a `Channel` instance. `aiochannel` will automatically use the currently running event loop.

affects: >=1.0.0

Errors

Common errors & fixes

TypeError: object Channel cannot be awaited

Attempting to call an asynchronous method (e.g., `put`, `get`) on a Channel object without `await` inside an `async def` function, or calling an `async def` function as if it were a regular function.

fix

Ensure all calls to `async` methods or `async def` functions are prefixed with `await`. For top-level execution, use `asyncio.run()`.

aiochannel.ChannelClosed: Channel is closed

Attempting to `put()` an item into a channel after `channel.close()` has been called, or attempting to `get()` from a channel that has been closed and is already empty.

fix

Design your producer logic to close the channel only when no more items will be sent. Consumers should handle the `ChannelClosed` exception, often by breaking out of a loop, or use `async for` which handles this implicitly.

Upgrade

Version history

1.3.0latest on PyPI · released Dec 9, 2024

Audit

Dependencies

No dependency data recorded yet.

Agent activity

60 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/tbug/aiochannel ↗

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