Skip to content
Connect > Client Libraries

Using the SQLAlchemy pyami-3forge Dialect

pyami.db.sqlalchemy_ami is a SQLAlchemy dialect for 3forge Enterprise. It lets you drive an AmiCore server with SQLAlchemy Core and the ORM, and plug into the wider ecosystem (pandas, Alembic-style reflection, etc.). Under the hood it routes through pyami.db.ami_dbapi (a PEP 249 / DB-API 2.0 wrapper) which talks to the native pyami.rt.ami_client extension over the AMI JDBC port.

1
2
3
4
5
6
7
8
9
SQLAlchemy Core / ORM
        │
   AmiDialect (this module)        ── translates SQL + DDL to AMI SQL
        │
   pyami.db.ami_dbapi             ── PEP 249 driver
        │
   pyami.rt.ami_client            ── native extension, JDBC wire protocol
        │
     3forge server

For the broader API and the non-SQLAlchemy layers, see the user guide.

Install & Register

The dialect ships inside pyami-3forge:

1
pip install pyami-3forge

It registers itself automatically through the sqlalchemy.dialects entry point (declared in pyproject.toml), so once the package is installed the ami+jdbc:// URL scheme just works: no manual step.

If you run from a source checkout without installing the package, register it by hand once per process:

1
2
from sqlalchemy.dialects import registry
registry.register("ami.jdbc", "pyami.db.sqlalchemy_ami.base", "AmiDialect")

Connection URL

1
ami+jdbc://<user>:<password>@<host>:<port>[?use_ssl=true&timeout_ms=60000&limit=10000]
Query parameter Default Description
use_ssl true SSL/TLS for the JDBC connection. Pass false/0/no to use plain TCP.
timeout_ms 60000 Query timeout in milliseconds.
limit 10000 Maximum rows returned per query.

Parameter names matter. Query parameters are forwarded verbatim as keyword arguments to the DB-API connect(), whose names are use_ssl, timeout_ms, and limit. Use timeout_ms, not timeout (an unknown keyword raises TypeError at connect time).

SSL is on by default, so a bare URL connects over SSL:

1
2
3
4
5
from sqlalchemy import create_engine

engine = create_engine("ami+jdbc://<user>:<password>@localhost:3280")
# plain TCP, custom timeout/limit:
engine = create_engine("ami+jdbc://<user>:<password>@localhost:3280?use_ssl=false&timeout_ms=30000&limit=5000")

For backup JDBC servers (failover), connect through the DB-API layer or the realtime client directly: see DB-API 2.0.

AMI SQL Differences You Will Notice

The dialect translates most standard SQLAlchemy constructs to AMI SQL automatically (see Operator & Function Translation). But when you write raw text() you must use AMI syntax yourself:

Standard SQL AMI SQL
= (equality) ==
'string' "string" (double quotes)
LIKE 'p%' ~~ "/^p.*/"
SELECT DISTINCT … SELECT … GROUP BY …
LIMIT n OFFSET m LIMIT m, n
CAST(x AS INT) (INT)(x)
NOW() timestamp()

~~ is a filter match, not LIKE. Its wildcards are *, for any run of characters, and ., for any single character; % and _ are ordinary characters. By default the match is partial and case-insensitive, so ~~ "p%" looks for the literal text p%. A pattern written between slashes is a regular expression: ~~ "/^p.*/" is anchored and case-sensitive, which is how LIKE 'p%' behaves.

Tables are created PUBLIC by default (the DDL compiler injects the PUBLIC prefix if you do not specify one). AMI CREATE TABLE does not accept PRIMARY KEY / UNIQUE / NOT NULL / FOREIGN KEY clauses. The dialect omits PRIMARY KEY, NOT NULL, and FOREIGN KEY, and replaces the primary key and foreign keys with separate CREATE INDEX ... USE CONSTRAINT=... statements (see DDL); NOT NULL is dropped with no replacement. UNIQUE is not omitted: a column declared with unique=True, or a UniqueConstraint, is written into the CREATE TABLE statement, which the Center rejects. Declare uniqueness instead with an Index that has unique=True or info={"use_constraint": "UNIQUE"}.

Core: Raw SQL

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from sqlalchemy import create_engine, text

engine = create_engine("ami+jdbc://<user>:<password>@localhost:3280")

with engine.connect() as conn:
    conn.execute(text("DROP PUBLIC TABLE IF EXISTS greet"))
    conn.execute(text("CREATE PUBLIC TABLE IF NOT EXISTS greet (msg STRING)"))
    conn.execute(text('INSERT INTO greet (msg) VALUES ("Hello, AMI!")'))
    row = conn.execute(text("SELECT msg FROM greet")).fetchone()
    print(row[0])

Bind parameters use the named paramstyle (:name):

1
2
3
4
5
6
7
8
with engine.connect() as conn:
    conn.execute(text("CREATE PUBLIC TABLE IF NOT EXISTS t (x INT, y STRING)"))
    conn.execute(
        text("INSERT INTO t (x, y) VALUES (:x, :y)"),
        [{"x": 1, "y": "a"}, {"x": 2, "y": "b"}],     # executemany
    )
    row = conn.execute(text("SELECT x, y FROM t WHERE x == :x"), {"x": 2}).fetchone()
    print(row)

commit()/rollback() are no-ops (Enterprise has no transactions). Each statement is applied as it executes; do not rely on rollback semantics. See Limitations.

Core: Tables & Expressions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from sqlalchemy import Table, Column, Integer, String, MetaData, select

meta = MetaData()
users = Table(
    "users", meta,
    Column("id", Integer),
    Column("name", String),
    prefixes=["PUBLIC"],        # optional — PUBLIC is added automatically if omitted
)

with engine.connect() as conn:
    meta.drop_all(conn, checkfirst=True)
    meta.create_all(conn)
    conn.execute(users.insert(), [{"id": 1, "name": "alice"}, {"id": 2, "name": "bob"}])
    rows = conn.execute(select(users).where(users.c.id == 2)).fetchall()
    print(rows)

users.c.id == 2 compiles to AMI SQL's id == :id_1; you write normal SQLAlchemy and the dialect emits AMI operators.

ORM

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

engine = create_engine("ami+jdbc://<user>:<password>@localhost:3280")

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    __table_args__ = {"prefixes": ["PUBLIC"]}

    # PK -> emitted as CREATE INDEX ... USE CONSTRAINT="PRIMARY"; autogen requests
    # a server-generated incrementing id.
    id: Mapped[int] = mapped_column(primary_key=True, info={"autogen": "INC"})
    name: Mapped[str] = mapped_column(String(30))

Base.metadata.drop_all(engine, checkfirst=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add_all([User(id=0, name="alice"), User(id=1, name="bob")])
    session.commit()
    for u in session.query(User).order_by(User.id):
        print(u.id, u.name)

DDL, Constraints & Indexes

Because AMI CREATE TABLE rejects standard constraint clauses, the dialect:

  • Omits PRIMARY KEY / NOT NULL / FOREIGN KEY from the column/table DDL.
  • Registers a Table.after_create hook that creates indexes for the primary key and foreign keys only:
  • primary key → CREATE INDEX idx_<table>_pk ON <table>(...) USE CONSTRAINT="PRIMARY"
  • foreign keys → CREATE INDEX idx_<table>_fk ON <table>(...) USE CONSTRAINT="NONE"

NOT NULL has no replacement. UNIQUE is not omitted, so declare it with an Index (below) rather than with unique=True on a column.

For a single-column primary key this is automatic: declaring a primary_key=True column is enough. The hook has 2 gaps, and neither raises an error:

  • Composite primary key: gets no index, and the table's foreign-key index is then skipped as well. Declare the primary-key index yourself with an Index and info={"use_constraint": "PRIMARY"}.
  • Several foreign keys: only one is indexed, and which one is not fixed. Declare an Index for each foreign key you need indexed.

Index constraint: choose the constraint kind explicitly on an Index:

1
2
from sqlalchemy import Index
idx = Index("by_symbol", table.c.symbol, info={"use_constraint": "UNIQUE"})  # NONE | UNIQUE | PRIMARY

Per-column index modifiers: (HASH, SORT, SERIES) 2 equivalent ways:

1
2
3
4
5
# (a) on the Index, keyed by column name:
Index("by_symbol", table.c.symbol, info={"ami_columns": {"symbol": "HASH"}})

# (b) on the Column itself:
Column("symbol", String, info={"ami_index": "SORT"})

Autogenerated values: (AUTOGEN="INC"|"RAND") on the primary-key column (propagates to the generated PK index) or directly on an index:

1
2
3
id: Mapped[int] = mapped_column(primary_key=True, info={"autogen": "INC"})
# or:
Index("rand_idx", table.c.id, info={"autogen": "RAND"})

Indexes are emitted as CREATE INDEX IF NOT EXISTS <name> ON <table>(<cols [MODIFIER]>) [USE CONSTRAINT="…"] [AUTOGEN="…"].

Operator & Function Translation

The statement compiler rewrites these for you (so the same SQLAlchemy expression works against AmiCore):

SQLAlchemy AMI SQL emitted
col == value col == :value
col != value col != :value
col.like("x%") col ~~ :value
col.notlike("x%") NOT (col ~~ :value)
select(...).distinct() SELECT … GROUP BY …
select(...).limit(n).offset(m) … LIMIT :m, :n
cast(col, Integer) (INT)(col)
func.now() timestamp()

like, ilike, startswith, endswith, contains, and their negations all go through ~~. The dialect converts the pattern to an anchored regular expression before binding it, with % becoming .* and _ becoming .: ilike and its variants are case-insensitive, and the others are case-sensitive. A negation is emitted as NOT (col ~~ :value), not !~, because the Center's !~ matches every row when given a regular expression. The pattern must be a value: like against another column cannot be translated and raises a compile error.

String literals are quoted with double quotes ("...") and identifiers with backticks (`...`), matching AMI SQL, where a double-quoted token is always a string literal.

Type Mapping

When the dialect emits DDL, SQLAlchemy types map to AMI types as follows:

SQLAlchemy type AMI type (DDL)
Integer INT
SmallInteger SHORT
BigInteger LONG
Float DOUBLE
REAL FLOAT
String / Unicode / Text STRING
CHAR CHAR
Boolean BOOLEAN
DateTime UTC
BINARY Binary
LargeBinary Binary

For reflection (reading types back from the server), AMI type names map to SQLAlchemy types via the dialect's ischema_names: e.g. Double→Float, Long/BigDecimal→ BigInteger, Short→SmallInteger, UTC/UTCN→DateTime, UUID→a UUID type, Complex→a complex type.

pandas Integration

Read straight into a DataFrame through a SQLAlchemy connection:

1
2
3
4
5
6
import pandas as pd

with engine.connect() as conn:
    df = pd.read_sql_table("user_account", con=conn)        # whole table
    df2 = pd.read_sql_query("SELECT name FROM user_account WHERE id == 1", con=conn)
print(df)

(For a lighter-weight query-to-DataFrame path that skips SQLAlchemy entirely, the native client's sql() and the pure-Python pyjdbc client both return DataFrame-ready output: see the user guide.)

Reflection / Inspection

inspect(engine) works against AmiCore's system tables (__TABLE, __COLUMN, __INDEX):

1
2
3
4
5
6
from sqlalchemy import inspect

insp = inspect(engine)
print(insp.get_table_names())
print(insp.get_columns("user_account"))
print(insp.get_indexes("user_account"))

has_table() and MetaData(...).reflect() are likewise supported, so checkfirst=True on create_all/drop_all does the right thing.

Limitations

  • No transactions: commit() and rollback() are no-ops, so statements take effect as they execute (skip_autocommit_rollback is set on the dialect).
  • One result set per statement: from a multi-result AmiCore response, only the first result set is returned.
  • JDBC protocol only: the dialect does not cover publish/subscribe or Center feeds, so use pyami.rt.ami_client directly for those.

Troubleshooting

Symptom Cause / fix
TypeError: ... unexpected keyword argument 'timeout' You used ?timeout= in the URL. The correct parameter is ?timeout_ms=.
Can't load plugin: sqlalchemy.dialects:ami.jdbc Package not installed (entry point missing). Either pip install pyami-3forge, or registry.register("ami.jdbc", "pyami.db.sqlalchemy_ami.base", "AmiDialect").
Could not parse SQLAlchemy URL from given URL string The URL is malformed, e.g: ami+jdbc// with the colon missing. Check it against the connection URL format.
SSL handshake / connection failures Server is not on SSL, or wrong port. Add ?use_ssl=false or correct the port.
commit() "does not persist a rollback" Expected: there are no transactions; every statement is immediately applied.
Syntax error on raw text() SQL You used standard SQL (=, '…', LIKE). Use AMI syntax (==, "…", ~~).
CREATE TABLE rejected for constraints A UNIQUE clause reached the DDL, from unique=True on a column or a UniqueConstraint, or PRIMARY KEY/UNIQUE was written inline in raw DDL. Declare primary keys with primary_key=True and uniqueness with an Index, or write CREATE INDEX ... USE CONSTRAINT=... yourself.

See also: the user guide (full API, DB-API layer, AMI SQL notes) and the Jupyter guide (interactive use, pandas, live feeds).