Handle Query
Register a handler for incoming KubeMQ queries and return data responses using the Python SDK.
Overview
A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.
Registering a handler by iterating subscribe_to_queries opens a subscription; the broker delivers every matching QueryReceived to your loop as it arrives. The handler builds a QueryResponse carrying the original query_received back to the broker, so the answer routes to the specific caller blocked waiting, and sets body with the real result before calling send_response.
Gotchas: if the handler never sends a response, the caller blocks until its own timeout_in_seconds elapses and fails with a timeout, not a fast error. An exception inside the handler doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same handler loop, slow handler code delays every other in-flight caller.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Handle query — subscribe and respond to incoming queries with data."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
QueriesSubscription,
QueryMessage,
QueryReceived,
QueryResponse,
)
async def main() -> None:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-queries-handle-query-client",
) as client:
# Subscribe to queries
token = AsyncCancellationToken()
async def query_handler() -> None:
async for query in client.subscribe_to_queries(
subscription=QueriesSubscription(
channel="python-queries.handle-query",
on_receive_query_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
"""Handle incoming query and send response with data."""
try:
body = query.body.decode("utf-8")
print(f"Handling query: Id={query.id}, Body={body}")
# Process the query (simulate data lookup)
result_data = f"Result for: {body}"
# Send response back with data
await client.send_response(
QueryResponse(
query_received=query,
is_executed=True,
body=result_data.encode(),
)
)
print(f" Response sent with data: {result_data}")
except Exception as e:
print(f" Error handling query: {e}")
handler_task = asyncio.create_task(query_handler())
await asyncio.sleep(1)
print("Listening for queries on 'python-queries.handle-query'...")
# Send a test query
response = await client.send_query(
QueryMessage(
channel="python-queries.handle-query",
body=b"fetch user profile",
timeout_in_seconds=10,
)
)
print(f"Query result: executed={response.is_executed}, body={response.body}")
await asyncio.sleep(1)
token.cancel()
handler_task.cancel()
try:
await handler_task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
subscribe_to_queries yields each QueryReceived with id, channel, body, metadata, and tags. The responder must call send_response(QueryResponse(query_received=query, is_executed=True, body=...)) with the result data — without body, the sender receives a response with an empty payload. Errors in the handler are caught and logged; an unhandled exception would kill the subscriber task silently, leaving the sender waiting until its timeout expires.
Related
Was this page helpful?