Picopyn - Picodata Python driver

Picopyn is a Python package for working with the distributed Picodata database.

Version compatibility:

Features

• Connection pooling with configurable pool size
• Optional automatic node discovery
• Pluggable load-balancing strategies
• Shard-aware routing
• Automatic pool membership reconciliation, including rebalancing a full pool across the nodes
• Restricting pool connections to replicaset leaders or replicas (instance_role)
• Temporary ejection of failing instances from pool selection
• Changing pool settings at runtime (e.g. max_size, topology_update_interval)
• Asynchronous API based on asyncpg
• Synchronous DB-API 2.0-compatible API based on psycopg

Navigation

• Installation
• Quickstart
• Development
  • Docs
  • Lint
  • Test
  • Debug
• Benchmark

Installation

pip install picopyn

Or from source:

git clone https://git.picodata.io/core/drivers/picopyn.git
cd picopyn
pip install -e .

Quickstart

Async:

import asyncio
from picopyn.asynchronous import Pool

async def main():
    # create and open pool to the picodata cluster
    pool = Pool(dsn="postgresql://admin:pass@localhost:5432", enable_discovery=True)
    await pool.open()

    # execute DDL operations
    await pool.execute('''
        CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id) OPTION (TIMEOUT = 3.0);
    ''')

    # execute DML/DQL operations
    await pool.execute('INSERT INTO "warehouse" VALUES ($1::int, $2::varchar)', 1, "test")
    rows = await pool.fetch('SELECT * FROM "warehouse"')
    print(rows)

    await pool.close()

asyncio.run(main())

Sync:

from picopyn.synchronous import connect

# create and connect to the picodata cluster
with connect("postgresql://admin:pass@localhost:5432") as conn:
    cur = conn.cursor()

    # execute DDL operations
    cur.execute('''
        CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id) OPTION (TIMEOUT = 3.0);
    ''')

    # execute DML/DQL operations
    cur.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
    cur.execute('SELECT * FROM "warehouse"')
    print(cur.fetchall())

Development

For development we use uv as package manager and docker compose for test environment.

To install development dependencies:

uv sync --extra test

Inside the test container, run the same command to keep dependencies up to date if the image is outdated:

make shell
# inside the container:
make install

Documentation

The documentation is written in Markdown and built with MkDocs using the Material theme. API reference is generated automatically from docstrings via mkdocstrings.

Published documentation is available at picopyn.readthedocs.io.

Source files are located in the docs/ directory. To work on docs locally, install the docs dependencies and start the live-reload server:

uv sync --extra docs
make doc

How to write code

We use several tools to ensure code style and type safety.

• ruff — code style, lint checks and automatic lint fixing
• mypy — static type checking
• black — code formatting

To check code style and static types:

make lint

To automatically fix formatting and style issues:

make fmt

How to test

We use docker compose for test environment.

Run the general test suite:

make test

or run SSL tests:

make test-ssl

This will:

1. Start required test containers (Picodata cluster and test-runner) using Docker Compose

2. Execute tests using pytest

How to debug

Do not forget to run the environment via make env

For debugging purposes:

1. Open a bash shell in the test container:

make shell

2. For interactive Python (with asyncio support) run inside of container:

python -m asyncio

3. To connect directly to Picodata:

picodata admin tmp/data/picodata-1-1/admin.sock

Benchmark

Benchmark instructions and usage examples are available here.