Skip to content
Connect > Client Libraries

Python client

The pyami package. It has four layers; pick the one that matches your task.

Layer Import Use it for
Native realtime client import pyami.rt.ami_client as ac Subscribing, publishing, commands, and SQL over one client
DB-API 2.0 from pyami.db.ami_dbapi import ami_dbapi as db Standard Python database access (connect/cursor/execute)
SQLAlchemy dialect create_engine("ami+jdbc://…") ORM, reflection, and pandas.read_sql
Pure-Python JDBC pyami.pyjdbc A native-free JDBC client, the only layer that runs without Python 3.14, but plaintext-only

Only pyami.pyjdbc runs without the compiled extension. The native client is the extension, and the DB-API and SQLAlchemy layers both import it internally, so all three require Python 3.14. If you are not on 3.14, pyami.pyjdbc is the only client that runs, but it is in the source repository only, and it does not support TLS (see below).

Installing

The client is distributed through your 3forge deployment or support channel, it is not on a public package index, so a plain pip install pyami-3forge against the open internet will not find it. Once you have access to the package (or your deployment's internal index), install it by name:

1
pip install pyami-3forge

The package requires Python 3.14 or newer and pulls in sqlalchemy and pandas.

The whole native-backed stack needs Python 3.14, including the DB-API and SQLAlchemy layers

The native client ships as a compiled cp314 extension. The DB-API and SQLAlchemy layers both import that native module internally, so they inherit the 3.14 requirement too, they are not a way to avoid it. The only layer that runs without the native extension is pyami.pyjdbc (see below), and it is plaintext-only. If you are not on Python 3.14, none of the native, DB-API, or SQLAlchemy paths will import.

The native client

Connect

Create a cluster, fill in the configs your program uses, and add it. Each config is (port, use_ssl, …); the JDBC and Center configs also take a username and password.

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

node = ac.AmiCluster("N1", "localhost")
node.jdbc_config   = ac.AmiDatabaseConfig("3280", True, "demo", "demo123", 60000, 10000)
node.center_config = ac.AmiSubscriptionConfig("3270", True)
node.relay_config  = ac.AmiPublicationConfig("3289", True, "python-client",
                                             ac.Options.ENABLE_SEND_SEQNUM)

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

AmiDatabaseConfig arguments are port, use_ssl, username, password, query_timeout_ms, query_limit. add_cluster(cluster, sync=False), pass sync=True to block until connected.

Run SQL

1
2
3
4
5
6
7
8
result = client.sql("N1", "SELECT SYM, bid FROM QUOTES")   # -> {"SYM": [...], "bid": [...]}
df = pd.DataFrame(result)

client.sql("N1", 'INSERT INTO QUOTES (SYM, bid) VALUES ("IBM", 123.4)')   # -> {"rows_affected": 1}
client.sql("N1", 'UPDATE QUOTES SET bid=123.6 WHERE SYM == "IBM"')        # -> {"rows_affected": 1}
client.sql("N1", "DROP TABLE QUOTES")                                     # DDL -> {}

client.sql_async("N1", "SELECT * FROM QUOTES", on_result)  # callback receives the same dict

The result is a column-oriented dict, and the shape tells you what ran:

  • SELECT → {col: [values]}.
  • INSERT / UPDATE / DELETE → {"rows_affected": n}.
  • DDL (CREATE / DROP) → an empty dict {}, that is success, not a failure.
  • error → {"error": "..."}.

WHERE uses the expression syntax, equality is == and string literals use double quotes.

Subscribe

Implement the callbacks you care about on any object and attach it, then subscribe by table name.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
class MyListener:
    def on_connect(self, client, name): ...
    def on_subscribe(self, client, name, tables): ...
    def on_update(self, client, name, update):
        # update.action: 1=add, 2=update, 3=delete
        fields = dict(zip(update.field_names, update.row_data))
        key = update.id            # merge on id, object_id is empty on subscription updates
        # also: update.table_name
        ...

client.add_client_listener(MyListener())
client.subscribe("N1", ["QUOTES", "ORDERS"])

The update object exposes action, table_name, id, object_id, and the parallel field_names / row_data. An update (action 2) carries only the changed columns. Merge keyed by id: the stable numeric row identity. object_id is the Relay-publish key and is empty on subscription updates for SQL-created rows, so do not merge on it.

Python does not have unsubscribe or get_table_schema

The native Python client binds subscribe but not unsubscribe or get_table_schema, those exist only in the C++ client. To stop receiving updates, call remove_cluster (see Disconnect and teardown) or drop the connection.

subscribe takes table names only. Python has no server-side WHERE/column form, the CenterSubscription object is C++-only. In Python, subscribe to the whole table and filter in your callback, or use the C++ client when you need the Center to filter.

There is no end-of-snapshot callback. on_subscribe fires when the subscription reaches the server, before the initial rows, which arrive afterward as on_update with action 1 (add). No callback announces "the initial load is complete." Pass sync=True to subscribe to block while the subscription is established; otherwise treat the initial add rows themselves as the snapshot.

Publish

1
2
3
4
5
6
7
client.publish("N1", "QUOTES", "IBM", [("SYM", "IBM"), ("bid", 123.4), ("ask", 123.5)])

objects = [{"id": "MSFT", "SYM": "MSFT", "bid": 410.1},
           {"id": "AAPL", "SYM": "AAPL", "bid": 189.2}]
client.publish_batch("N1", "QUOTES", objects)

client.remove("N1", "IBM", "QUOTES")   # delete the object

publish(node, object_type, id, fields, seqnum=0, expires_on_ms=0) takes fields as a list of (name, value) tuples. publish_batch takes a list of dicts, or a pandas DataFrame, which it publishes row by row.

Bulk insert

For high-volume loads, pass a dict of column arrays (lists or numpy arrays), the binary path, no SQL text:

1
2
3
4
5
6
7
data = {"id": np.array([1, 2, 3], dtype=np.int32),
        "price": np.array([9.99, 4.50, 12.0]),
        "name": ["Apple", "Pear", "Mango"]}

client.sql_bulk_insert("N1", "fruits", data)          # -> {"rows_affected": 3}
client.sql_insert("N1", "fruits", data)               # INSERT
client.sql_replace("N1", "fruits", data, suffix="")   # REPLACE: the rows passed become the table contents; suffix = persist engine

sql_insert and sql_replace also accept a pandas DataFrame in place of the dict.

TLS and mutual TLS

Set the SSL options on the config's shard:

1
2
3
4
5
6
7
node.jdbc_config = ac.AmiDatabaseConfig("3280", True, "demo", "demo123")
shards = node.jdbc_config.shards
shards[0].ssl = ac.AmiSslOptions(
    trusted_server_pub_file="keys/server.pub",   # pin the server
    client_cert_file="keys/client.crt.pem",      # present our cert (mutual TLS)
    client_key_file="keys/client.key.pem")
node.jdbc_config.shards = shards

See Configuration → TLS.

Disconnect and teardown

remove_cluster is the counterpart to add_cluster: it drops that cluster's JDBC, Center, and Relay connections and stops delivering its updates. It is also how you stop a subscription (Python has no unsubscribe).

1
2
client.remove_cluster("N1")               # remove_cluster(name, sync=False)
client.remove_cluster("N1", sync=True)    # block until the cluster is torn down

The process may not exit on its own

The native client runs its own I/O threads, and they can keep the interpreter alive after main returns, even once every cluster is removed. If your program hangs at exit, end it explicitly (for example os._exit(0)) once your work is done.

DB-API 2.0

A standard PEP-249 driver, the easiest way to run one-off queries.

1
2
3
4
5
6
7
from pyami.db.ami_dbapi import ami_dbapi as db

conn = db.connect(host="localhost", port=3280, username="demo", password="demo123")
cur = conn.cursor()
cur.execute("SELECT * FROM QUOTES")
rows = cur.fetchall()
conn.close()

connect(host, port=3280, username, password, timeout_ms=60000, limit=10000, use_ssl=True, failover_endpoints=None) (plus the SSL cert arguments). The paramstyle is named (:name placeholders). There are no transactions: commit and rollback do nothing.

SQLAlchemy and pandas

1
2
3
4
5
6
7
8
9
from sqlalchemy import create_engine
import pandas as pd

engine = create_engine("ami+jdbc://demo:demo123@localhost:3280")
# or: create_engine("ami+jdbc://demo:demo123@localhost:3280?use_ssl=false&timeout_ms=30000")

with engine.connect() as conn:
    df = pd.read_sql_table("QUOTES", con=conn)                 # whole table
    df2 = pd.read_sql_query('SELECT SYM, bid FROM QUOTES WHERE SYM == "IBM"', con=conn)

The dialect supports Core, ORM, and reflection. See the quant analytics use case.

List the tables

The native and DB-API layers have no dedicated "list tables" call. Use SQLAlchemy reflection to enumerate tables and their columns; it queries the same server, so the list matches what the native client sees:

1
2
3
4
5
from sqlalchemy import inspect

insp = inspect(engine)
insp.get_table_names()          # -> ['QUOTES', 'ORDERS', ...]
insp.get_columns("QUOTES")      # -> [{'name': 'SYM', 'type': ...}, ...]

Pure-Python JDBC (pyami.pyjdbc)

Not installed by pip

The pyami.pyjdbc layer is not included in the packaged client: pip install pyami-3forge does not install it, and importing it fails with ModuleNotFoundError. It exists in the source repository only.

pyami.pyjdbc is a self-contained JDBC client with no native extension: it is the only Python layer that runs on a Python older than 3.14. It connects, logs in, and runs a query straight to a pandas DataFrame:

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

conn = AmiSocketConn("localhost", 3280, "demo", "demo123")   # host, port, user, password
conn.connect()
df = conn.to_dataframe("SELECT SYM, bid, ask FROM QUOTES")
conn.disconnect()

AmiSocketConn(host, port, user, password, timeout=60000, limit=10000, failover_endpoints=None); connect(), to_dataframe(sql), disconnect().

pyjdbc is plaintext-only, it cannot reach a TLS-secured port

pyami.pyjdbc opens a plain TCP socket with no TLS. Against a server whose JDBC listener requires TLS or mutual TLS (the common secure configuration), the connection will hang or be rejected. Use it only against a non-TLS listener. For a secured server you need the native client (Python 3.14) or the C++ client, both of which do TLS, see Configuration → Connecting to a secured server.

Concurrency model

The client is callback- and thread-based, not asyncio-based. There is no awaitable API: sql_async takes a callback, and subscription events arrive on listener callbacks that run on the client's I/O threads. If you work in an asyncio program, marshal results from the callbacks onto your event loop yourself (for example with loop.call_soon_threadsafe).

What this does not support

  • No transactions. The DB-API driver's commit and rollback are no-ops; there is no unit of work.
  • No unsubscribe, get_table_schema, or CenterSubscription in Python. All three are C++ only. Python subscribe takes table names; stopping a subscription means removing the cluster or dropping the connection, and filtering happens in the callback rather than on the server.
  • No asyncio. The client is callback- and thread-based; sql_async takes a callback, not an awaitable.
  • No native layer below Python 3.14. The native client, DB-API, and SQLAlchemy layers all require it; only pyami.pyjdbc runs on an older Python, and it is plaintext-only (no TLS).
  • One JDBC request at a time per cluster. The SQL socket does not queue concurrent requests.

Verified live (build 32437)

The native client, DB-API, and SQLAlchemy paths on this page were exercised against a live instance over mutual TLS.