Send Query
Send a KubeMQ query and receive a data response from a handler using the Python SDK.
Overview
This tutorial builds the RPC half of KubeMQ's request/reply patterns: a query, where the caller awaits a handler's data payload instead of just a completion status. Reach for it whenever a caller needs an answer — fetching a record, running a lookup, or asking another service to compute a value on demand. You'll run a handler and a sender in the same process to see the full round trip.
The sender calls client.send_query with a QueryMessage and a timeout_in_seconds, then awaits a reply. client.subscribe_to_queries yields each incoming QueryReceived; the handler builds a QueryResponse(query_received=query, is_executed=True, body=...), and KubeMQ uses the request's correlation ID to route that reply back to the exact caller waiting on it.
Gotchas: the timeout must cover however long the handler takes to run — a slow handler raises KubeMQTimeoutError even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately, so startup order matters. body is raw bytes — encoding it is your application's job.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Send query — send a query and receive a response with data."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
KubeMQConnectionError,
KubeMQError,
KubeMQTimeoutError,
QueriesSubscription,
QueryMessage,
QueryReceived,
QueryResponse,
)
async def main() -> None:
try:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-queries-send-query-client",
) as client:
# Set up a query responder
token = AsyncCancellationToken()
async def query_handler() -> None:
async for query in client.subscribe_to_queries(
subscription=QueriesSubscription(
channel="python-queries.send-query",
on_receive_query_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"Responder received: {query.body.decode('utf-8')}")
await client.send_response(
QueryResponse(
query_received=query,
is_executed=True,
body=b"response data payload",
)
)
handler_task = asyncio.create_task(query_handler())
await asyncio.sleep(1)
# Send a query and get the response
response = await client.send_query(
QueryMessage(
channel="python-queries.send-query",
body=b"hello kubemq, please reply!",
timeout_in_seconds=10,
)
)
print(
f"Response: executed={response.is_executed}, "
f"body={response.body}, "
f"timestamp={response.timestamp}"
)
token.cancel()
handler_task.cancel()
try:
await handler_task
except asyncio.CancelledError:
pass
except KubeMQConnectionError as e:
print(f"Connection error: {e}")
except KubeMQTimeoutError as e:
print(f"Timeout error: {e}")
except KubeMQError as e:
print(f"KubeMQ error: {e}")
if __name__ == "__main__":
asyncio.run(main())
# Expected output:
# Responder received: hello kubemq, please reply!
# Response: executed=True, body=b'response data payload', timestamp=<timestamp>
How It Works
Unlike commands, send_query expects the responder to include data in the response body. The QueryResponse carries both is_executed=True and body=b"response data payload". send_query returns a QueryResponse with is_executed, body, timestamp, and optionally cache_hit (for cached queries). Queries are otherwise structurally identical to commands — same async generator subscription, same correlation-ID routing, same timeout_in_seconds behaviour.
Related
Was this page helpful?