Subscribe to ticking tables from a Python client

This guide shows how to receive live updates from a Deephaven table in an external Python application with the pydeephaven-ticking package. Use it to build a dashboard, an alerting service, a feed into a downstream system or ML pipeline, or any other application that reacts to table changes without running inside the Deephaven server.

A ticking table is one whose contents can change while the server runs — the server can add, remove, or modify rows at any time. The base pydeephaven package can only fetch a point-in-time snapshot of such a table, so keeping up with a ticking table means repeatedly re-fetching the whole thing and diffing it yourself. pydeephaven-ticking does that work for you: your code receives callbacks that contain just the rows that changed. Callbacks don't arrive for every individual change; see how the server batches them.

Note

This guide covers the client package, which runs outside the Deephaven server. Choose it when your code runs in a separate application or process. If your code already runs on the server — for example, in a query — use Listen to ticking tables instead; it's simpler there because there's no network or serialization involved.

Install the package

Install pydeephaven-ticking from PyPI with pip. Check the platform requirement first, because the prebuilt packages cover only one platform.

Important

pydeephaven-ticking wraps Deephaven's C++ client with Cython, and is published to PyPI as prebuilt packages for Linux on x86_64 only. No macOS, Windows, or Linux ARM packages are available. See the version matrix for details.

pip install pydeephaven-ticking

This also installs pydeephaven. Install the version that matches your server, for example pip install pydeephaven-ticking==42.8 for a 42.8 server.

On other platforms, run your client code in a Linux x86_64 container. For example, this starts a Python shell with the package installed:

docker run --rm -it --platform linux/amd64 python:3.12-slim sh -c 'pip install pydeephaven-ticking && python'

The --platform linux/amd64 flag matters on ARM computers, such as Macs with Apple silicon. Without it, Docker runs an ARM container, and pip can't find a pydeephaven-ticking package to install. With it, Docker emulates x86_64, which runs more slowly than a native container.

Inside a container, localhost refers to the container itself, not to your computer. Point the Session at the server in one of these ways:

Server on your computer, Docker Desktop on macOS or Windows: connect with Session(host="host.docker.internal", port=10000).
Server on your computer, Linux: add --network host to docker run, then connect to localhost.
Server in another container: add --network container:<server-container-name> to docker run, then connect to localhost.

You don't import pydeephaven_ticking directly. When it's installed, pydeephaven exposes four extra names: listen, TableListener, TableUpdate, and TableListenerHandle.

from pydeephaven import Session, TableListener, TableUpdate, listen

If that import fails, run import pydeephaven_ticking.table_listener to see why. pydeephaven only adds the four names when it can import that submodule — which in turn loads the compiled _core extension — and it hides the reason when it can't. Plain import pydeephaven_ticking doesn't reproduce the failure, since the package's __init__.py does no work of its own. The direct submodule import shows the real error — for example, the package isn't installed in the active environment, or its compiled library can't load on your platform.

Note

The examples in this guide create a Session with no auth_type, which defaults to anonymous authentication. A Deephaven server started from the default configuration instead requires pre-shared key authentication; for such a server, pass auth_type="io.deephaven.authentication.psk.PskAuthenticationHandler" and auth_token="<your key>" to Session, as shown in Capture Python client tables.

The simplest listener is a function that takes one argument — a TableUpdate:

import time
from pydeephaven import Session, TableUpdate, listen


def on_update(update: TableUpdate) -> None:
    added = update.added()
    if added:
        print(added["X"].to_pylist())


session = Session(host="localhost", port=10000)
try:
    table = session.time_table("PT1s").update(["X = ii"])
    handle = listen(table, on_update)
    handle.start()  # Subscribe and start receiving updates on a background thread.
    try:
        time.sleep(10)  # The main thread is free to do other work.
    finally:
        handle.stop()  # Unsubscribe and wait for the background thread to exit.
finally:
    session.close()

listen takes a Table reference and a listener, and returns a TableListenerHandle. Nothing happens until you call start. The try/finally blocks make sure the handle is stopped and the session closed even if your code fails or you press Ctrl+C; see Stop the subscription. The table can be any table reference — one you create with the session, or an existing server table opened with open_table.

listen raises ValueError if a function listener or error callback doesn't take exactly one argument, if you pass on_error together with a TableListener object (use the object's on_error method instead), or if the listener is neither a function nor a TableListener.

For listeners that keep state or need custom error handling, subclass TableListener. You must implement on_update; on_error is optional.

import time

from pydeephaven import Session, TableListener, TableUpdate, listen


class PrintingListener(TableListener):
    def on_update(self, update: TableUpdate) -> None:
        for kind, cols in [
            ("removed", update.removed()),
            ("added", update.added()),
            ("modified", update.modified()),
        ]:
            if cols:
                print(kind, {name: arr.to_pylist() for name, arr in cols.items()})

    def on_error(self, error: Exception) -> None:
        print(f"Subscription failed: {error}")


session = Session(host="localhost", port=10000)
try:
    table = session.time_table("PT1s").update(["X = ii"])
    handle = listen(table, PrintingListener())
    handle.start()
    try:
        time.sleep(10)
    finally:
        handle.stop()
finally:
    session.close()

Read the changes in a table update

The server batches changes: it waits at least one update interval, one second by default, between propagations of ongoing changes to your subscription. Configure that interval with the server-side -Dbarrage.minUpdateInterval setting (in milliseconds); see What is Barrage? for details. If nothing changed during an interval, no callback arrives. Don't rely on the interval to space out your callbacks, though: a large table's initial snapshot can arrive in size-limited rounds that the server sends back to back, and one propagation can deliver more than one update, so your listener can run several times within one interval.

Each call to on_update receives a TableUpdate that describes the coalesced, net row changes since the previous call — not a log of every intermediate change. Blink tables are the exception; see the last point below. Every accessor returns a dict that maps column names to PyArrow arrays. The arrays in one dict all have the same length, and position i in each array belongs to the same row. If a category has no rows in this update, the accessor returns an empty dict.

Method	Returns
`added`	Rows added in this update.
`removed`	Rows removed in this update, with the values they had as of the previous update.
`modified`	Rows modified in this update, with their new values.
`modified_prev`	The same rows as `modified`, in the same order, with the values before the change.

Keep these points in mind:

The initial contents arrive as one or more updates. When the subscription starts, every row currently in the table arrives as an addition. For a large table, the server can send the snapshot in size-limited rounds, and the client reports a separate update for each round, so the first callback may not contain the whole table, and ordinary changes can arrive between rounds. A row you already received can also be sent again in a later round, and the update reports it as removed and then added with its new data, so apply removals before additions. Don't treat the first callback as a fully initialized local copy. If the table is empty at that moment, the first update has no rows. Blink tables are the exception: the server always sends an empty initial snapshot for a blink table, even if rows are present at subscription time. Don't treat that first, empty update as the blink table's current contents — the first real rows arrive in later updates.
Modified rows include every requested column. A row is reported as modified if any of its columns changed, and modified returns all the columns you ask for, not just the ones that changed. This means you can always read a key column alongside the changed values.
modified and modified_prev line up by position. Use them together to see how each row changed — for example, to compute a price move.
For tables that aren't blink tables, one update can combine several changes to the same row. The server merges all the changes from an update interval, so each row appears at most once per category:
- A row modified several times appears once in modified, with its latest values. modified_prev holds its values from before the first of those changes.
- A row added and then modified appears only in added, with its latest values. It isn't also in modified.
- A row added and then removed doesn't appear at all.
- A row removed and then added again appears in both removed and added. Apply removals before additions so that the row ends up present.
- A row that changed and was then removed during the interval appears in removed with its values from before those changes, not its last values before it was removed.
Blink tables aren't coalesced. The server keeps every row that blinked during an update interval, so a row that appeared and disappeared within the interval still arrives in added, and the rows delivered by the previous update arrive in removed.

Request only the columns you need

Every accessor takes an optional cols argument — a single column name or a list of names. By default, it returns all columns.

def on_update(update: TableUpdate) -> None:
    prices = update.modified(["Sym", "Price"])
    if prices:
        print(dict(zip(prices["Sym"].to_pylist(), prices["Price"].to_pylist())))

Process large updates in chunks

Each accessor has a _chunks variant — added_chunks, removed_chunks, modified_chunks, and modified_prev_chunks — that returns a generator instead of one dict. Each chunk holds at most chunk_size rows. Use these to limit memory use when an update, such as the initial snapshot of a large table, contains many rows.

def on_update(update: TableUpdate) -> None:
    for chunk in update.added_chunks(10_000, ["Sym", "Price"]):
        # Each chunk maps column names to PyArrow arrays of up to 10,000 rows.
        process(chunk)

Data types

Deephaven column types map to these PyArrow types. Null values arrive as PyArrow nulls, which become None when you call to_pylist.

Deephaven type	PyArrow type
`byte`	`int8`
`short`	`int16`
`int`	`int32`
`long`	`int64`
`float`	`float32`
`double`	`float64`
`char`	`uint16` (a number, not a string)
`boolean`	`bool`
`String`	`string`
`Instant`	`timestamp("ns", "UTC")`
`ZonedDateTime`	`timestamp("ns", "UTC")` (the zone is discarded)
`LocalDate`	`date64`
`LocalTime`	`time64("ns")`
Arrays/vectors of a type above (except `byte[]`)	`list` of the PyArrow type

Other column types, such as enums or other custom objects, are supported too: the server stringifies their values, so they arrive as PyArrow string.

A smaller set of types fail instead: the server encodes byte[], BigDecimal, BigInteger, and Schema columns as Arrow binary, and Duration, Period, and PeriodDuration columns as Arrow duration/interval types — none of which this package's schema conversion supports.

Because start converts the whole schema before it launches the background thread, a column with one of these types makes start raise synchronously, the same way byte[] does. Nested arrays and vectors (an array of arrays, or a vector of vectors) aren't supported either, for the same reason — only one level of list wrapping is supported. Drop or convert those columns on the server first — for example with view — before you subscribe.

A subscription always covers every row and every column of the table you pass to listen. Shape the table on the server first, so that only the data you need crosses the network:

where to keep only relevant rows
view to keep only relevant columns
tail to keep only the most recent rows
last_by to keep only the latest row per key

trades = session.open_table("trades")
latest = trades.where("Exchange = `NYSE`").view(["Sym", "Price", "Size"]).last_by("Sym")
handle = listen(latest, on_update)

Threading

Each TableListenerHandle runs its own background thread. Your listener's methods run on that thread, one update at a time, in order. Keep the following in mind:

Don't block in on_update. Updates for this subscription wait while your callback runs. Hand slow work, such as database writes, to another thread or a queue.
Protect shared state. If the main thread reads data that on_update writes, guard it with a threading.Lock.
You can call stop from inside on_update. This ends the subscription after the current callback returns, which is useful for stopping when a condition is met.

Handle errors

If on_update raises an exception while the subscription is running, the subscription ends and the listener's on_error method runs with the exception. No more updates arrive after that. The default on_error prints the error.

on_error only covers failures during active background processing, after start has launched the background thread:

Synchronous setup failures raise directly from start. Opening the connection and writing the subscription request happen on the calling thread, before the background thread exists, so errors there propagate out of start as normal exceptions instead of reaching on_error. A table with a column type the package doesn't support fails this way; see Data types.
Background stream failures reach on_error. Once the background thread is running, failures while reading the stream — including a subscription the server rejects — are caught there and passed to on_error.
A clean end of the stream is not an error. The background thread treats it as normal completion and stops quietly without calling on_error.

When your listener is a function, pass an error callback as the third argument to listen:

def on_error(error: Exception) -> None:
    print(f"Subscription failed: {error}")


handle = listen(table, on_update, on_error)

To recover, create a new handle and call start again. If you'd rather keep the subscription running after a bad update, catch exceptions inside on_update.

Stop the subscription

Call stop when you're done. Called from outside the listener thread, it cancels the subscription and waits for the background thread to exit, so no further callbacks run after it returns. Called from inside on_update (as described above), it only marks the subscription cancelled and returns immediately without waiting — the current callback keeps running, and the background thread exits on its own once that callback finishes.

The listener thread isn't a daemon thread, so if you skip stop — for example, because Ctrl+C interrupts your code before it gets there — the thread keeps running and your program doesn't exit. Close the session after stopping its handles. Use try/finally so that cleanup happens even when your program fails or is interrupted:

handle = listen(table, on_update)
try:
    handle.start()
    try:
        run_application()
    finally:
        handle.stop()
finally:
    session.close()

Example: keep a live local copy of a table

This example keeps a local dict in sync with a ticking table keyed by symbol and prints alerts for large price moves. It applies removals first, then additions, then modifications.

import threading
import time

from pydeephaven import Session, TableListener, TableUpdate, listen

COLS = ["Sym", "Price", "Size"]


def rows(cols: dict) -> list[tuple]:
    """Turns a dict of column name to PyArrow array into a list of row tuples."""
    return list(zip(*(cols[c].to_pylist() for c in COLS))) if cols else []


class QuoteBook(TableListener):
    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.quotes: dict[str, tuple[float, int]] = {}

    def on_update(self, update: TableUpdate) -> None:
        removed = update.removed("Sym")
        added = update.added(COLS)
        prev, curr = update.modified_prev(COLS), update.modified(COLS)

        with self.lock:
            for sym in removed.get("Sym", []):
                self.quotes.pop(sym.as_py(), None)
            for sym, price, size in rows(added):
                self.quotes[sym] = (price, size)
            for (sym, old_price, _), (_, price, size) in zip(rows(prev), rows(curr)):
                self.quotes[sym] = (price, size)
                if None not in (old_price, price) and abs(price - old_price) >= 4.0:
                    print(f"ALERT {sym}: {old_price:.2f} -> {price:.2f}")

    def snapshot(self) -> dict[str, tuple[float, int]]:
        with self.lock:
            return dict(self.quotes)


session = Session(host="localhost", port=10000)
try:
    quotes = (
        session.time_table("PT0.25s")
        .update(
            [
                "Sym = `SYM` + randomInt(0, 5)",
                "Price = 100.0 + randomGaussian(0.0, 5.0)",
                "Size = randomInt(1, 1000)",
            ]
        )
        .last_by("Sym")
    )

    book = QuoteBook()
    handle = listen(quotes, book)
    handle.start()
    try:
        for _ in range(10):
            time.sleep(3)
            print(book.snapshot())
    finally:
        handle.stop()
finally:
    session.close()

The server table uses last_by, so each new symbol arrives as an added row and later trades for that symbol arrive as modified rows. If a symbol's first trade and later trades fall in the same update interval, only an added row arrives, with the latest trade. The listener handles that case without special code because it stores every added row. The listener reads Sym with every modification even though only Price and Size change, which is what lets it find the right entry in the dict.