ParticleDB for Python

import particledb

with particledb.connect() as db:
    db.kv_set(b"session:123", b"active")
    assert db.kv_get(b"session:123") == b"active"

Start a local server

A server started without options requires authentication but has no password set, so start it in one of these two ways. For local development on a trusted machine, turn authentication off and keep every listener on loopback:

particledb start --no-auth --data-dir ./pdb-dev \
  --http-addr 127.0.0.1:7173 --pg-addr 127.0.0.1:5432

To require a password, create the login role from the command line. The password file holds only the password; keep it readable by you alone. SCRAM authenticates without TLS; use TLS for remote access.

particledb start --data-dir ./pdb-dev \
  --http-addr 127.0.0.1:7173 --pg-addr 127.0.0.1:5432 \
  --auth-method scram-sha-256 --pg-username app --pg-password-file ./pdb-password

The SDK connects to localhost:7173 by default. With authentication on, pass the same user and password:

import os
import particledb

with particledb.connect(user="app", password=os.environ["PARTICLEDB_PASSWORD"]) as db:
    assert db.scalar("SELECT 42") == 42

The default address is localhost:7173. Core queries, transactions and streaming use the standard library. Python 3.11 or newer is required.

Install the release build

This release is not published to PyPI yet. Build a wheel with python -m build from this directory and install the resulting particledb-0.2.0-py3-none-any.whl in your application. Optional extras are arrow, pandas, polars, sqlalchemy and grpc. See Python SDK documentation for the matching server.

Remote connections

import os
import particledb

with particledb.connect(
    host="db.example.com",
    user="app",
    password=os.environ["PARTICLEDB_PASSWORD"],
    sslmode="verify-full",
    sslrootcert="ca.pem",
) as db:
    print(db.scalar("SELECT 42"))

verify-full checks the certificate and hostname. Omit sslrootcert to use system roots. TLS uses TCP and refuses plaintext fallback. sslcert and sslkey accept PEM client certificate and key paths. require encrypts without hostname verification; use verify-full for verified remote access.

Queries and transactions

with particledb.connect() as db:
    db.execute("CREATE TABLE items (id BIGINT PRIMARY KEY, name TEXT)")
    db.execute("INSERT INTO items VALUES ($1, $2)", [1, "first"])
    assert db.query("SELECT name FROM items WHERE id = $1", [1]) == [{"name": "first"}]
    with db.transaction() as tx:
        tx.execute("UPDATE items SET name = $1 WHERE id = $2", ["updated", 1])

Parameters are bound separately from SQL. The transaction context commits on successful exit and rolls back on an exception. SQL failures include SQLSTATE; transport failures do not replay a request that may have run. Synchronous write_timeout defaults to 30 seconds and bounds socket writes and waits for another writer. A write timeout closes the connection because the request outcome may be unknown. Set it to None for an unlimited wait. statement_timeout separately bounds response waits and server cancellation.

Malformed connection URLs raise ValueError in the native API and InterfaceError in DB-API and asyncio. Error messages omit the connection string and do not chain the original URL parser exception.

Use stream(sql, parameters) for bounded row iteration and close an iterator early when abandoning a result.

Values decode to Python int, float, bool, str, bytes, Decimal, date, time, datetime, UUID, lists and dictionaries. SQL NULL is None. Decimals preserve their scale. Aware datetimes represent instants; naive datetimes represent timestamps without a time zone.

DB-API and asyncio

from particledb import dbapi

with dbapi.connect() as db:
    with db.cursor() as cursor:
        cursor.execute("SELECT %s AS answer", (42,))
        assert cursor.fetchone() == (42,)

The DB-API defaults to %s parameters. Its connection context manages a transaction. Use autocommit=True for one transaction per statement. Cursors provide fetchone, fetchmany, fetchall, description and executemany. A named cursor streams in bounded memory.

from particledb import aio

async def read_answer():
    async with await aio.connect() as db:
        async with db.cursor() as cursor:
            await cursor.execute("SELECT %s AS answer", (42,))
            return await cursor.fetchone()

The async API follows the DB-API shapes with awaited I/O and async context managers. Cancellation is sent to the server. A connection is reused only when the server confirms the statement's outcome; otherwise it is closed. Do not share an open transaction between unrelated tasks.

DataFrames and SQLAlchemy

Install the corresponding optional extra. DB-API cursors provide fetch_arrow, fetch_df, fetch_polars and fetch_numpy. bulk_insert(table, rows) accepts supported tuples, mappings, generators and DataFrames. Use a transaction for an all-or-nothing application operation.

from sqlalchemy import create_engine, text

engine = create_engine("particledb+native://localhost:7173")
with engine.connect() as db:
    assert db.execute(text("SELECT :answer"), {"answer": 42}).scalar() == 42
engine.dispose()

The dialect supports parameterized SQLAlchemy Core statements, transactions, reflection and streaming. See SUPPORT.md for tested behavior, server limitations and the separate compliance suite.

Validate

PDB_NATIVE_PORT=PORT PDB_TLS_PORT=TLS_PORT PDB_TLS_CA=ca.pem \
  PDB_PG_DSN=postgresql://user@localhost:PG_PORT/sdk_python \
  python -m pytest -q -ra

Use disposable databases. Live tests require the named listeners. SQLAlchemy compliance runs separately against a fresh server directory; it is not part of a plain pytest invocation. See Python SDK documentation.

MIT licensed. See LICENSE.