Using XTDB from Python

In Python, you can talk to a running XTDB node using the standard psycopg tooling, taking advantage of XTDB’s Postgres wire-compatibility.

import asyncio
import psycopg as pg

DB_PARAMS = {
    "host": "localhost",
    "port": 5432
}

async def insert_trades(conn, trades):
    query = """
    INSERT INTO trades (_id, name, quantity) VALUES (%s, %s, %s)
    """

    async with conn.cursor() as cur:
        for trade in trades:
            trade_values = (trade["_id"], trade["name"], trade["quantity"])
            await cur.execute(query, trade_values)

async def get_trades_over(conn, quantity):
    query = """
    SELECT * FROM trades WHERE quantity > %s
    """
    async with conn.cursor() as cur:
        await cur.execute(query, (quantity,))
        return await cur.fetchall()

async def main():
    trades = [
        {"_id": 1, "name": "Trade1", "quantity": 1001},
        {"_id": 2, "name": "Trade2", "quantity": 15},
        {"_id": 3, "name": "Trade3", "quantity": 200},
    ]

    try:
        async with await pg.AsyncConnection.connect(**DB_PARAMS, autocommit=True) as conn:
            # required for now https://github.com/xtdb/xtdb/issues/3589
            conn.adapters.register_dumper(str, pg.types.string.StrDumperVarchar)
            await insert_trades(conn, trades)
            print("Trades inserted successfully")

            result = await get_trades_over(conn, 100)
            print(result)
    except Exception as error:
        print(f"Error occurred: {error}")

if __name__ == "__main__":
    asyncio.run(main())

Arrow-native access via ADBC

For Arrow-native workloads (query results as Arrow batches, bulk-ingesting an Arrow table in one round trip, zero-copy hand-off to pandas / polars / DuckDB), XTDB also exposes ADBC over its FlightSQL listener. In Python that’s the adbc_driver_flightsql driver:

pip install adbc-driver-flightsql pyarrow

import adbc_driver_flightsql.dbapi as flight_sql

with flight_sql.connect("grpc://localhost:9832") as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")
        print(cur.fetch_arrow_table())

See the ADBC tutorial for an end-to-end Python walkthrough and the ADBC reference for the supported surface.

Examples

For more examples and tests, see the XTDB driver-examples repository, which contains comprehensive test suites demonstrating various features and use cases.