Skip to content
Connect > Client Libraries

Using the pyami AMI Client in a Jupyter Notebook

This guide shows how to use pyami-3forge interactively in Jupyter / JupyterLab, from one-off SQL queries that render as DataFrames, to live, self-updating dashboards driven by 3forge Enterprise realtime and Center feeds.

It assumes you already know the API basics; for the full reference see the user guide. For the SQLAlchemy dialect specifically, see the SQLAlchemy guide.

Install in a Notebook

Use the %pip magic so the package installs into the kernel's interpreter (not whatever pip happens to be on PATH):

1
2
%pip install pyami-3forge
%pip install panel            # only needed for the live-dashboard sections

Then restart the kernel (Kernel -> Restart) so the freshly installed native extension is importable.

Python 3.14 required. The native pyami.rt.ami_client extension is compiled for CPython 3.14 (wheels are tagged cp314). If %pip install reports "no matching distribution", your kernel is on a different Python: check with:

1
import sys; print(sys.version)

Confirm the extension imports:

1
2
from pyami.rt import ami_client as ac
print("pyami ready")

Query into a DataFrame (JDBC)

The simplest notebook use is a synchronous SQL query rendered as a pandas DataFrame. sql() returns a column-oriented dict that pd.DataFrame(...) consumes directly:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import pandas as pd
import pyami.rt.ami_client as ac

node = ac.AmiCluster("jdbc", "localhost")
node.jdbc_config = ac.AmiDatabaseConfig("3280", True, "<user>", "<password>", 60000, 10000)

client = ac.AmiClient()
client.add_cluster(node, sync=True)        # blocks this cell until connected

result = client.sql("jdbc", 'SELECT name, size FROM planets WHERE name == "Earth"')
df = pd.DataFrame(result)                   # JupyterLab renders this as a table
df

Remember AMI SQL's quirks (== for equality, "double quotes" for string literals): see Run SQL.

If you only need read-only query-to-DataFrame access and would rather not load the native extension, a pure-Python client works on any platform and any Python 3. It is in the source repository only, and pip install pyami-3forge does not install it:

1
2
3
4
5
6
7
from pyami.pyjdbc import AmiSocketConn

conn = AmiSocketConn(host="localhost", port=3280, user="<user>", password="<password>")
conn.connect()
df = conn.to_dataframe("SELECT * FROM pdemo")
conn.disconnect()
df

You can also go through the SQLAlchemy dialect and let pandas drive it: see pandas integration.

The Golden Rule: Callbacks Run on Background Threads

Every listener callback (on_update, on_connect, on_message_received, …) and every sql_async result callback is invoked on an internal C++ I/O thread, not on the notebook's event loop. This has 2 practical consequences in a notebook:

  1. Never touch a widget (Panel/ipywidgets) directly from a callback. Widget state must be mutated on the front-end's IOLoop thread. Marshal the update instead:
1
2
3
4
5
6
7
8
9
import tornado.ioloop

ioloop = tornado.ioloop.IOLoop.current()   # capture on the MAIN thread, in a cell

class MyListener:
    def on_update(self, client, name, update):
        # runs on a C++ thread — do NOT call widget.patch(...) here
        value = ...                         # extract plain data
        ioloop.add_callback(lambda: my_widget.patch(value))   # safe: runs on IOLoop
  1. Never call back into the client re-entrantly from a callback. Calling subscribe() (or other blocking client methods) directly inside on_connect can deadlock or crash the C++ I/O thread. Enqueue the request and perform it from your own loop instead:
1
2
3
4
5
6
7
8
9
import collections
pending = collections.deque()

class MyListener(ac.AmiClientListener):
    def on_connect(self, client, name):
        if name.endswith("-center"):        # names arrive as "<node>-center"/"-relay"
            pending.append(name[:-len("-center")])

# drain `pending` and call client.subscribe(node, [...]) from a timer/IOLoop callback

Both patterns come from the multiplex demo, which installs with pyami-3forge (see Runnable Demos).

If you only read plain Python data (no widgets, no re-entrant client calls), a callback can do its work inline, just keep it short and non-blocking, because it holds up the I/O thread while it runs.

Async SQL Without Blocking the Kernel

sql() blocks the cell until the server responds. For long queries, or to keep an interactive cell responsive, use sql_async() and marshal the result to a display:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import pandas as pd, tornado.ioloop
from IPython.display import display

ioloop = tornado.ioloop.IOLoop.current()

def on_result(result: dict):
    # runs on the C++ network thread — hop to the IOLoop before displaying
    ioloop.add_callback(lambda: display(pd.DataFrame(result)))

client.sql_async("jdbc", "SELECT * FROM pdemo", on_result)

Live Dashboards with Panel

Panel + its Tabulator widget is the path of least resistance for a live table in a notebook. Load the extension once at the top of the notebook:

1
2
import panel as pn
pn.extension("tabulator")

The recommended pattern (used by the demos) is:

  1. Build a Tabulator from an initial DataFrame and display it.
  2. On each incoming AmiCore update, compute the changed cells.
  3. Marshal a Tabulator.patch({...}) call to the IOLoop: patching individual cells is far cheaper than reassigning .value.
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import pandas as pd, panel as pn, tornado.ioloop
pn.extension("tabulator")

df = pd.DataFrame({"Symbol": ["BTC", "ETH"], "Price": [0.0, 0.0]})
table = pn.widgets.Tabulator(df, show_index=False)
row_of = {sym: i for i, sym in enumerate(df["Symbol"])}

ioloop = tornado.ioloop.IOLoop.current()

class PriceListener:
    def on_update(self, client, name, update):
        try:
            fields = dict(zip(update.field_names, update.row_data))
            sym, price = fields["symbol"], float(fields["price"])
        except (UnicodeDecodeError, KeyError, ValueError):
            return                              # skip rows we can't decode/parse
        i = row_of.get(sym)
        if i is not None:
            ioloop.add_callback(lambda: table.patch({"Price": [(i, price)]}))

client.add_client_listener(PriceListener())
client.subscribe("center", ["crypto"])         # async; snapshot + deltas flow to on_update

table                                            # display the live widget

For richer styling (conditional colors, flash-on-change, multi-source columns), study demo.crypto_ticker_nb and demo.multiplex_crypto_ticker, which install with the package (see Runnable Demos).

Subscribing to a Center Table Live

The Center delivers a snapshot followed by row-level deltas. In a notebook, connect with sync=True so the cell blocks until the connection is up, then subscribe:

1
2
3
4
5
6
7
8
node = ac.AmiCluster("center", "localhost")
node.center_config = ac.AmiSubscriptionConfig("3270", True, "<user>", "<password>")

client = ac.AmiClient()
client.add_cluster(node, sync=True)

# sync=True waits for the initial snapshot to arrive before the cell returns
client.subscribe("center", ["OrdersD", "Trades"], sync=True)

update.action distinguishes the delta type (1=ADD, 2=UPD, 3=DEL). An UPD carries only the columns that changed, so key rows on update.id and merge each UPD into the row already held: see Subscriptions. Guard access to string fields that may carry binary payloads: see Subscribe.

Cell Re-runs, Lifecycle, and Cleanup

Notebooks invite re-running cells. Each AmiClient owns background threads and sockets, so re-running a "connect" cell without cleaning up leaks connections and stacks duplicate callbacks. 2 habits keep this sane:

Tear down before reconnecting. Make the connect cell idempotent:

1
2
3
4
5
6
7
try:
    client.remove_cluster("center")      # disconnect the previous run, if any
except NameError:
    pass

client = ac.AmiClient()
client.add_cluster(node, sync=True)

Guard background loops with a running flag so a re-run does not start a second timer (the demos' CryptoTicker.start() does exactly this):

1
2
if not getattr(ticker, "_running", False):
    ticker.start(router)

When you are done, stop your producer and drop the clusters:

1
2
ticker.stop()                 # cancels timers, joins in-flight ticks
client.remove_cluster("center")

Gotchas

Symptom Cause / fix
%pip install → "no matching distribution" Kernel is not CPython 3.14. The wheels are cp314-tagged. Check sys.version.
Widget never updates from a feed You mutated the widget from a callback thread. Marshal via ioloop.add_callback(...).
Kernel hangs / crashes after connect You called subscribe() (or another blocking client call) directly inside on_connect. Enqueue it and run from the IOLoop instead.
Duplicate / doubled updates after re-running cells A previous AmiClient is still alive. remove_cluster(...) before reconnecting.
Cell hangs forever on add_cluster(..., sync=True) Cannot reach the server, or wrong port/SSL. Try use_ssl=False, verify host/port.
UnicodeDecodeError reading update.table_name/fields Binary payload in a string field. Wrap access in try/except UnicodeDecodeError.
Callbacks make the UI sluggish Heavy work on the I/O thread. Keep callbacks short; offload to your own queue/loop.

Runnable Demos

Two notebook-oriented demos install with pyami-3forge, as modules of the demo package alongside pyami. They are the most complete worked examples of the patterns on this page:

Module Shows
demo.crypto_ticker_nb A Panel Tabulator live ticker driven by an asyncio loop, with no connection to a Center: the display and styling pattern on its own.
demo.multiplex_crypto_ticker A complete integration: publish to one Relay, receive the Center's incremental updates on a listener, forward transformed updates to a second instance, and update Panel safely from callback threads.

A notebook cell wires the multiplex demo in one call:

1
2
3
4
5
6
7
8
from demo.multiplex_crypto_ticker import CryptoTicker, create_default_multiplex_stack

ticker = CryptoTicker()
client, router, listener = create_default_multiplex_stack(ticker)
ticker.start(router)
ticker.app          # display the live dashboard
# ...later...
ticker.stop()

By default the multiplex demo connects to 2 instances on localhost, with Relay ports 3289 and 4289 and Center ports 3271 and 4271, and uses the table crypto. Pass hosts, relay_ports, center_ports, or table_name to create_default_multiplex_stack to point it elsewhere.