List Channels
List all active KubeMQ channels on the server and inspect their stats using the Python SDK admin API.
Overview
Listing channels turns the broker into a discoverable inventory instead of a black box — instead of hardcoding channel names everywhere, you ask the server what actually exists right now. That's exactly what monitoring dashboards, cleanup scripts, and "did my deployment create the channels it should have" checks need. It's read-only and has no effect on message flow, so it's safe to run against production at any time.
Under the hood, each client's list_*_channels_async method — list_events_channels_async, list_queues_channels_async, list_commands_channels_async, and so on — queries channels of one type. PubSubClient covers events and events-store, QueuesClient covers queues, CQClient covers commands and queries; no single call spans all types. Each result is a ChannelInfo with name, type, and activity stats.
Gotchas: results are scoped to whichever client you call from — asking PubSubClient about queues channels isn't possible, so audits that need the full picture must query all three clients. A channel type with no matches returns an empty list rather than raising, so check length instead of wrapping every call in a try/except for control flow. And the activity stats are a snapshot at query time — a channel shown active can go idle immediately after.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: List channels — list all channels by type."""
from __future__ import annotations
import asyncio
from kubemq import CQClient
from kubemq import PubSubClient
from kubemq import QueuesClient
async def main() -> None:
# List events and events store channels
async with PubSubClient(
address="localhost:50000",
client_id="python-management-list-channels-client",
) as client:
try:
events_channels = await client.list_events_channels_async()
print(f"Events channels: {events_channels}")
except Exception as e:
print(f"Error listing events channels: {e}")
try:
events_store_channels = await client.list_events_store_channels_async()
print(f"Events store channels: {events_store_channels}")
except Exception as e:
print(f"Error listing events store channels: {e}")
# List queues channels
async with QueuesClient(
address="localhost:50000",
client_id="python-management-list-channels-client",
) as client:
try:
queues_channels = await client.list_queues_channels_async()
print(f"Queues channels: {queues_channels}")
except Exception as e:
print(f"Error listing queues channels: {e}")
# List CQ channels
async with CQClient(
address="localhost:50000",
client_id="python-management-list-channels-client",
) as client:
try:
commands_channels = await client.list_commands_channels_async()
print(f"Commands channels: {commands_channels}")
except Exception as e:
print(f"Error listing commands channels: {e}")
try:
queries_channels = await client.list_queries_channels_async()
print(f"Queries channels: {queries_channels}")
except Exception as e:
print(f"Error listing queries channels: {e}")
if __name__ == "__main__":
asyncio.run(main())
How It Works
list_*_channels_async returns a list of ChannelInfo objects, each describing a channel's name, type, and activity stats. Results are scoped to the channel type each client manages: PubSubClient lists events and events-store channels, QueuesClient lists queues channels, and CQClient lists commands and queries channels. If no channels of a type exist the call returns an empty list rather than raising.
Related
Was this page helpful?