Queries and Subscriptions

Every Q exposes three ways to talk to q:

Python	Node.js	q message	Returns
sync(expr, *args)	sync(expression, ...args)	Synchronous request	The decoded reply. q errors are raised.
asyn(expr, *args)	asyn(expression, ...args)	Asynchronous message	None in Python. Node.js resolves with no value once the bytes are written. q sends no reply.
receive()	receive()	None. It reads the next message the server sends.	The decoded message

All three are blocking calls in Python and return promises in Node.js.

String queries and functional calls

With only an expression, q evaluates the string:

import xqdb

conn = xqdb.Q("localhost", 1800)
conn.sync("trade:([] sym:`AAPL`MSFT`AAPL; price:10.5 11.0 10.75)")
frame = conn.sync("select avg price by sym from trade")
conn.disconnect()

With arguments, xqdb sends the general list (expression; arg1; ...) and q applies the function to the arguments. The expression can name a function or be a lambda literal. An expression that starts with { and ends with } is sent as a q lambda. Each call takes at most eight arguments. More raise TypeError in Python and reject with XQDB_CONVERSION in Node.js.

Python

from datetime import date

import xqdb

conn = xqdb.Q("localhost", 1800)
conn.sync("vwap:{[t;s] exec size wavg price from t where sym=s}")
conn.sync("trade:([] sym:`AAPL`AAPL; price:10.0 11.0; size:100 300)")

print(conn.sync("vwap", "trade", "AAPL"))  # 10.75: named function
print(conn.sync("{x+y}", 1, 2))  # 3: lambda literal
print(conn.sync("{[d] d+1}", date(2024, 1, 31)))  # 2024-02-01: a datetime.date
conn.disconnect()

Node.js

import { Q, XqdbDate } from "@xbbg/xqdb";

const q = await Q.connect({ host: "localhost", port: 1800 });
try {
  await q.sync("vwap:{[t;s] exec size wavg price from t where sym=s}");
  await q.sync("trade:([] sym:`AAPL`AAPL; price:10.0 11.0; size:100 300)");

  console.log(await q.sync("vwap", "trade", "AAPL")); // 10.75: named function
  console.log(await q.sync("{x+y}", 1n, 2n)); // 3n: lambda literal, long arguments
  console.log(await q.sync("{[d] d+1}", new XqdbDate("2024-01-31"))); // XqdbDate, value "2024-02-01"
} finally {
  await q.disconnect();
}

Arguments convert by the type mapping. The conversions used most often:

You pass	Python	Node.js	q receives
Text	str	string	symbol
Bytes	bytes	Buffer, Uint8Array	string (char vector)
Integer	int	bigint	long
Floating point	float	number	float
List	list, tuple	array	general list
Dictionary	dict with str keys	plain object	symbol-keyed dictionary
Table	Any eager DataFrame	Arrow Table	table
Typed list	Any eager Series, pyarrow.ChunkedArray	Arrow Vector	typed list

q operators and lambdas as values

Use XqdbQOperator and XqdbQLambda to pass a q primitive or a lambda as an argument. They are also what xqdb returns when q sends back a primitive or a function.

Python

import xqdb
from xqdb import XqdbQLambda, XqdbQOperator

conn = xqdb.Q("localhost", 1800)
apply = "{[op;a;b] .[op;(a;b)]}"
print(conn.sync(apply, XqdbQOperator.PLUS, 1, 2))  # 3
print(conn.sync(apply, XqdbQOperator("*"), 3, 4))  # 12
print(conn.sync(apply, XqdbQLambda("{x-y}"), 10, 4))  # 6

scoped = XqdbQLambda("{x+y}", "analytics")  # runs in namespace .analytics
conn.disconnect()

Node.js

import { Q, XqdbQLambda, XqdbQOperator } from "@xbbg/xqdb";

const q = await Q.connect({ host: "localhost", port: 1800 });
try {
  const apply = "{[op;a;b] .[op;(a;b)]}";
  console.log(await q.sync(apply, XqdbQOperator.PLUS, 1n, 2n)); // 3n
  console.log(await q.sync(apply, new XqdbQOperator("*"), 3n, 4n)); // 12n
  console.log(await q.sync(apply, new XqdbQLambda("{x-y}"), 10n, 4n)); // 6n

  const scoped = new XqdbQLambda("{x+y}", "analytics"); // runs in namespace .analytics
  console.log(scoped.context);
} finally {
  await q.disconnect();
}

• XqdbQOperator(name) takes a q primitive name such as "+", "*:", "in", or "'". XqdbQOperator.PLUS is predefined. Any other name raises XqdbError (Python) or rejects with XQDB_CONVERSION (Node.js). The type mapping lists every supported name.
• XqdbQLambda(source, context="") takes the source text, which must be enclosed in braces and may start with k), plus an optional namespace. Give the namespace without its leading dot: "analytics" means .analytics, and a value that starts with . is rejected. NUL characters are rejected in both fields.
• Lambda source is executable q code. Build it only from trusted input.

Asynchronous messages

asyn() writes the message and returns. q sends no reply, so a q error in an asynchronous message is never reported to the client. q processes the messages from one connection in the order they arrive, so a later sync() on the same connection runs after every earlier asyn().

import pyarrow as pa
import xqdb

conn = xqdb.Q("localhost", 1800)
conn.sync("trade:([] sym:`symbol$(); price:`float$())")
rows = pa.table({"sym": pa.array(["AAPL"]).dictionary_encode(), "price": [10.5]})
for _ in range(3):
    conn.asyn("insert", "trade", rows)
print(conn.sync("count trade"))  # 3: runs after the three inserts
conn.disconnect()

Subscriptions

receive() returns the next message the server pushes on the connection, for example a kdb+tick tickerplant update or a message a q callback sends with neg .z.w. A standard tickerplant update arrives as ("upd", table_name, rows). Python returns it as a tuple with a DataFrame in it. Node.js returns an array with an Arrow Table in it.

Give every subscription its own connection:

• sync() sends a request and returns the next message on the socket. On a subscribed connection, that message can be a pushed update instead of the reply.
• A pending receive() holds the connection until a message arrives or the socket times out.
• A dropped connection loses the subscription. Reconnect, then subscribe again.

Python

import pyarrow as pa
import xqdb

feed = xqdb.Q("tickerplant.example.com", 5010)  # timeout=0: wait indefinitely
symbols = pa.chunked_array([pa.array(["AAPL", "MSFT"]).dictionary_encode()])


def subscribe() -> None:
    feed.sync(".u.sub", "trade", symbols)  # `trade; `AAPL`MSFT


subscribe()
while True:
    try:
        function, table, rows = feed.receive()
    except xqdb.XqdbIOError:
        feed.connect()
        subscribe()
        continue
    print(function, table, len(rows))

Node.js

import { Q, XqdbIOError, type XqdbValue } from "@xbbg/xqdb";
import { Table } from "apache-arrow";

// Keep the socket timeout above the longest quiet period of the feed (max 24 h).
const feed = await Q.connect({ host: "tickerplant.example.com", port: 5010, timeout: 3_600_000 });
const subscribe = () => feed.sync(".u.sub", "trade", "");

function consume(message: XqdbValue): void {
  if (!Array.isArray(message)) return;
  const [, table, rows] = message;
  if (rows instanceof Table) console.log(table, rows.numRows);
}

try {
  await subscribe();
  for (;;) {
    try {
      consume(await feed.receive());
    } catch (error) {
      if (!(error instanceof XqdbIOError) || error.code !== "XQDB_IO") throw error;
      await feed.connect(); // the timeout closed the socket: reconnect and resubscribe
      await subscribe();
    }
  }
} finally {
  await feed.disconnect();
}

Passing "" subscribes to every symbol, because an empty string becomes the null symbol `. To subscribe to specific symbols, pass a symbol list. In Python, use a dictionary-encoded pyarrow.ChunkedArray or a categorical Series, as in the example above. In Node.js, use a dictionary-encoded Arrow Vector, such as one built with vectorFromArray(["AAPL", "MSFT"]), which dictionary-encodes strings by default.

Keep retries=0 on Python subscription connections. A retried receive() reconnects but does not subscribe again. See Timeouts and Retries.