Sync

`picopyn.synchronous`

Sync API based on psycopg for connection management and pools.

`connect = Connection.connect` `module-attribute`

`Connection`

A representation of a database session.

Parameters:

Name	Type	Description	Default
`dsn`		The data source name (e.g., "postgresql://user:pass@host:port" or "postgresql://user:pass@host1:port1,host2:port2") for the picodata node.	required
`**connect_kwargs`		Additional keyword arguments to pass to psycopg.connect(). SSL can be configured either via DSN query params or via kwargs such as `sslmode`, `sslrootcert`, `sslcert`, `sslkey`.	required

Note

SSL should be configured using the appropriate parameters in a single source: either via DSN or by providing libpq SSL parameters in kwargs.

See Also

DB-API 2.0 Connection Objects specification.

Examples:

Simple usage:

conn = Connection.connect("postgresql://admin:pass@localhost:5432")
cur = conn.cursor()
cur.execute("SELECT 1")
conn.close()

Multi-host DSN string:

conn = Connection.connect("postgresql://admin:pass@host1:5432,host2:5432")

SSL (kwargs-based):

conn = Connection.connect(
    "postgresql://admin:pass@localhost:5432",
    sslmode="verify-ca",
    sslrootcert="/path/to/ca.crt",
    sslcert="/path/to/client.crt",
    sslkey="/path/to/client.key",
)

SSL (DSN query params):

conn = Connection.connect(
    "postgresql://admin:pass@localhost:5432/db"
    "?sslmode=verify-ca&sslrootcert=/path/to/ca.crt"
    "&sslcert=/path/to/client.crt&sslkey=/path/to/client.key"
)

Source code in picopyn/synchronous/connection.py

class Connection:
    """A representation of a database session.

    Args:
        dsn: The data source name (e.g., "postgresql://user:pass@host:port" or
            "postgresql://user:pass@host1:port1,host2:port2") for the picodata node.
        **connect_kwargs: Additional keyword arguments to pass to psycopg.connect().
            SSL can be configured either via DSN query params or via kwargs such as
            `sslmode`, `sslrootcert`, `sslcert`, `sslkey`.

    Note:
        SSL should be configured using the appropriate parameters in a single source:
        either via DSN or by providing libpq SSL parameters in kwargs.

    See Also:
        DB-API 2.0 [Connection Objects](https://peps.python.org/pep-0249/#connection-objects) specification.

    Examples:
        Simple usage:
        ```python
        conn = Connection.connect("postgresql://admin:pass@localhost:5432")
        cur = conn.cursor()
        cur.execute("SELECT 1")
        conn.close()
        ```

        Multi-host DSN string:

        ```python
        conn = Connection.connect("postgresql://admin:pass@host1:5432,host2:5432")
        ```

        SSL (kwargs-based):

        ```python
        conn = Connection.connect(
            "postgresql://admin:pass@localhost:5432",
            sslmode="verify-ca",
            sslrootcert="/path/to/ca.crt",
            sslcert="/path/to/client.crt",
            sslkey="/path/to/client.key",
        )
        ```

        SSL (DSN query params):

        ```python
        conn = Connection.connect(
            "postgresql://admin:pass@localhost:5432/db"
            "?sslmode=verify-ca&sslrootcert=/path/to/ca.crt"
            "&sslcert=/path/to/client.crt&sslkey=/path/to/client.key"
        )
        ```
    """

    dsn: str
    _connect_dsn: str
    _connect_kwargs: dict[str, Any]
    conn: psycopg.Connection[tuple[object, ...]] | None

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        raise TypeError(
            "Connection cannot be instantiated directly; use Connection.connect(dsn) instead."
        )

    def __repr__(self) -> str:
        # Explicit repr backed by the already-redacted `dsn`, so printing/logging
        # a Connection can never expose the password held in `_connect_dsn`.
        return f"Connection(dsn={self.dsn!r})"

    @classmethod
    def connect(cls, dsn: str, **connect_kwargs: Any) -> Self:
        """
        Create and return a new connection to Picodata.
        """
        if dsn is None:
            raise ValueError("dsn can not be None")

        validate_ssl_source(dsn, connect_kwargs)

        autocommit = connect_kwargs.pop("autocommit", None)
        if autocommit is not None and autocommit is not True:
            raise ValueError("autocommit=False is not supported")

        connection = cls.__new__(cls)
        # DSN with hidden password
        connection.dsn = _redact_dsn(dsn)
        # origin DSN to connect
        connection._connect_dsn = dsn
        connection._connect_kwargs = connect_kwargs
        connection.conn = None

        try:
            # With psycopg's default autocommit=False, even SELECT opens an implicit transaction.
            # See: https://www.psycopg.org/psycopg3/docs/basic/transactions.html#autocommit-transactions
            # Required until Picodata supports COMMIT.
            connection.conn = psycopg.connect(
                connection._connect_dsn,
                autocommit=True,
                **connection._connect_kwargs,
            )
        except Exception as error:
            raise RuntimeError(
                f"Failed to connect to picodata instance using DSN {connection.dsn}: {error}"
            ) from error

        return connection

    def close(self) -> None:
        """
        Close the connection gracefully.
        """
        if self.conn is None:
            return

        conn = self.conn

        try:
            conn.close()
        except Exception as e:
            raise RuntimeError(
                f"Failed to disconnect from picodata instance {self.dsn}: {e}"
            ) from e
        finally:
            self.conn = None

    def cursor(self) -> Cursor:
        """
        Return a new cursor object using the connection, which is used to
        manage the context of a fetch operation.

        Returns:
            A new Cursor bound to this connection.

        See [`Cursor`][picopyn.synchronous.cursor.Cursor] for full documentation.
        """
        self._ensure_open()

        assert self.conn is not None
        return Cursor(self.conn.cursor())

    def commit(self) -> None:
        """
        No-op for SQLAlchemy compatibility.

        DB-API defines this method as committing any pending transaction.
        """
        self._ensure_open()

    def rollback(self) -> None:
        """
        No-op for SQLAlchemy compatibility.

        DB-API defines this method as rolling back to the start of any pending transaction.
        """
        self._ensure_open()

    def _ensure_open(self) -> None:
        if self.conn is None or self.conn.closed:
            raise InterfaceError(
                "No active connection. Call .connect() before using the connection."
            )

`close()`

Close the connection gracefully.

Source code in picopyn/synchronous/connection.py

def close(self) -> None:
    """
    Close the connection gracefully.
    """
    if self.conn is None:
        return

    conn = self.conn

    try:
        conn.close()
    except Exception as e:
        raise RuntimeError(
            f"Failed to disconnect from picodata instance {self.dsn}: {e}"
        ) from e
    finally:
        self.conn = None

`commit()`

No-op for SQLAlchemy compatibility.

DB-API defines this method as committing any pending transaction.

Source code in picopyn/synchronous/connection.py

def commit(self) -> None:
    """
    No-op for SQLAlchemy compatibility.

    DB-API defines this method as committing any pending transaction.
    """
    self._ensure_open()

`connect(dsn, **connect_kwargs)` `classmethod`

Create and return a new connection to Picodata.

Source code in picopyn/synchronous/connection.py

@classmethod
def connect(cls, dsn: str, **connect_kwargs: Any) -> Self:
    """
    Create and return a new connection to Picodata.
    """
    if dsn is None:
        raise ValueError("dsn can not be None")

    validate_ssl_source(dsn, connect_kwargs)

    autocommit = connect_kwargs.pop("autocommit", None)
    if autocommit is not None and autocommit is not True:
        raise ValueError("autocommit=False is not supported")

    connection = cls.__new__(cls)
    # DSN with hidden password
    connection.dsn = _redact_dsn(dsn)
    # origin DSN to connect
    connection._connect_dsn = dsn
    connection._connect_kwargs = connect_kwargs
    connection.conn = None

    try:
        # With psycopg's default autocommit=False, even SELECT opens an implicit transaction.
        # See: https://www.psycopg.org/psycopg3/docs/basic/transactions.html#autocommit-transactions
        # Required until Picodata supports COMMIT.
        connection.conn = psycopg.connect(
            connection._connect_dsn,
            autocommit=True,
            **connection._connect_kwargs,
        )
    except Exception as error:
        raise RuntimeError(
            f"Failed to connect to picodata instance using DSN {connection.dsn}: {error}"
        ) from error

    return connection

`cursor()`

Return a new cursor object using the connection, which is used to manage the context of a fetch operation.

Returns:

Type	Description
`Cursor`	A new Cursor bound to this connection.

See Cursor for full documentation.

Source code in picopyn/synchronous/connection.py

def cursor(self) -> Cursor:
    """
    Return a new cursor object using the connection, which is used to
    manage the context of a fetch operation.

    Returns:
        A new Cursor bound to this connection.

    See [`Cursor`][picopyn.synchronous.cursor.Cursor] for full documentation.
    """
    self._ensure_open()

    assert self.conn is not None
    return Cursor(self.conn.cursor())

`rollback()`

No-op for SQLAlchemy compatibility.

DB-API defines this method as rolling back to the start of any pending transaction.

Source code in picopyn/synchronous/connection.py

def rollback(self) -> None:
    """
    No-op for SQLAlchemy compatibility.

    DB-API defines this method as rolling back to the start of any pending transaction.
    """
    self._ensure_open()

`Cursor`

DB-API cursor wrapper.

Supports query execution, cursor metadata and fetch methods.

See Also

DB-API 2.0 Cursor Objects specification.

Source code in picopyn/synchronous/cursor.py

class Cursor:
    """
    DB-API cursor wrapper.

    Supports query execution, cursor metadata and fetch methods.

    See Also:
        DB-API 2.0 [Cursor Objects](https://peps.python.org/pep-0249/#cursor-objects) specification.
    """

    def __init__(self, cursor: psycopg.Cursor[tuple[object, ...]]) -> None:
        self._cursor = cursor

    @property
    def description(self) -> Description | None:
        """
        Return DB-API column metadata for the current result set.

        Returns:
            A tuple of per-column metadata (name, type_code, display_size, internal_size, precision, scale, null_ok),
                or None if there is no result set.

        Examples:
            ```python
            cursor.execute('SELECT * FROM "warehouse"')
            [col[0] for col in cursor.description]
            # ['id', 'item']
            ```
        """
        self._ensure_open()

        description = self._cursor.description
        if description is None:
            return None

        return tuple(
            (
                column.name,
                column.type_code,
                column.display_size,
                column.internal_size,
                column.precision,
                column.scale,
                column.null_ok,
            )
            for column in description
        )

    @property
    def rowcount(self) -> int:
        """
        Return the number of rows produced or affected by the last operation.

        Returns:
            The row count, or -1 if it cannot be determined.
        """
        self._ensure_open()
        return self._cursor.rowcount

    @property
    def statusmessage(self) -> str | None:
        """
        Return the command status from the last operation.

        Returns:
            The status message reported by the database, or None if no command
            status is available.

        Examples:
            ```python
            cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
            cursor.statusmessage
            # 'INSERT 0 1'
            ```
        """
        self._ensure_open()
        return self._cursor.statusmessage

    def fetchone(self) -> tuple[object, ...] | None:
        """
        Fetch the next row from the current result set.

        Returns:
            The next row, or None if no more rows are available.

        Examples:
            ```python
            cursor.execute('SELECT * FROM "warehouse"')
            row = cursor.fetchone()
            # (1, 'test')
            ```
        """
        self._ensure_result_set()
        return self._cursor.fetchone()

    def fetchmany(self, size: int | None = None) -> list[tuple[object, ...]]:
        """
        Fetch the next set of rows from the current result set.

        Args:
            size: Maximum number of rows to fetch. If omitted, `arraysize` is used.

        Returns:
            A list of the fetched rows. Empty if no more rows are available.

        Examples:
            Use the previously configured `arraysize`:

            ```python
            cursor.arraysize = 100
            rows = cursor.fetchmany()
            ```

            Override the batch size for a single call:

            ```python
            rows = cursor.fetchmany(50)
            ```
        """
        self._ensure_result_set()

        if size is None:
            return self._cursor.fetchmany()
        return self._cursor.fetchmany(size)

    @property
    def arraysize(self) -> int:
        """
        Number of rows to fetch at a time when `fetchmany()` is called without `size`.

        Returns:
            The current arraysize.
        """
        self._ensure_open()
        return self._cursor.arraysize

    @arraysize.setter
    def arraysize(self, value: int) -> None:
        """
        Set the number of rows to fetch at a time with fetchmany().
        """
        self._ensure_open()
        self._cursor.arraysize = value

    def fetchall(self) -> list[tuple[object, ...]]:
        """
        Fetch all remaining rows from the current result set.

        Returns:
            A list of the remaining rows. Empty if no more rows are available.

        Examples:
            ```python
            cursor.execute('SELECT * FROM "warehouse"')
            rows = cursor.fetchall()
            # [(1, 'test')]
            ```
        """
        self._ensure_result_set()
        return self._cursor.fetchall()

    def execute(self, operation: QueryNoTemplate, parameters: Params | None = None) -> "Cursor":
        """
        Prepare and execute a database operation (query or command).

        Args:
            operation: database operation to be prepared.
            parameters: sequence or mapping that will be bound to variables in the operation.

        Returns:
            This cursor, to allow chaining (e.g. `cursor.execute(query).fetchall()`).

        Examples:
            DDL example:

            ```python
            ddl = 'CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id);'
            cursor.execute(ddl)
            ```

            DML example:

            ```python
            cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
            ```
        """
        self._ensure_open()

        try:
            self._cursor.execute(operation, parameters)
        except psycopg.Error as error:
            _raise_dbapi_error(error)

        return self

    def executemany(self, operation: QueryNoTemplate, seq_of_parameters: Iterable[Params]) -> None:
        """
        Prepare a database operation and execute it against all parameter sequences
        or mappings found in the sequence seq_of_parameters.

        Args:
            operation: database operation to be prepared.
            seq_of_parameters: all parameter sequences or mappings that will be bound to variables in the operation.

        Examples:
            ```python
            cursor.executemany(
                'INSERT INTO "warehouse" VALUES (%s, %s)',
                [(1, "test1"), (2, "test2")],
            )
            ```
        """
        self._ensure_open()
        try:
            self._cursor.executemany(operation, seq_of_parameters)
        except psycopg.Error as error:
            _raise_dbapi_error(error)

    def __iter__(self) -> Iterator[tuple[object, ...]]:
        """
        Iterate over the current result set.

        Returns:
            An iterator over the remaining rows.

        Examples:
            ```python
            cursor.execute("SELECT id, name FROM users")
            for row in cursor:
                print(row)
            ```
        """
        self._ensure_result_set()
        return iter(self._cursor)

    def close(self) -> None:
        """
        Close the cursor.
        """
        self._cursor.close()

    def _ensure_open(self) -> None:
        if self._cursor.closed:
            raise InterfaceError("Cursor is closed.")

    def _ensure_result_set(self) -> None:
        self._ensure_open()

        if self._cursor.description is None:
            raise ProgrammingError(
                "No result set is available. Execute a query that returns rows before fetching."
            )

`arraysize` `property` `writable`

Number of rows to fetch at a time when fetchmany() is called without size.

Returns:

Type	Description
`int`	The current arraysize.

`description` `property`

Return DB-API column metadata for the current result set.

Returns:

Type	Description
`Description \| None`	A tuple of per-column metadata (name, type_code, display_size, internal_size, precision, scale, null_ok), or None if there is no result set.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
[col[0] for col in cursor.description]
# ['id', 'item']

`rowcount` `property`

Return the number of rows produced or affected by the last operation.

Returns:

Type	Description
`int`	The row count, or -1 if it cannot be determined.

`statusmessage` `property`

Return the command status from the last operation.

Returns:

Type	Description
`str \| None`	The status message reported by the database, or None if no command
`str \| None`	status is available.

Examples:

cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
cursor.statusmessage
# 'INSERT 0 1'

`iter()`

Iterate over the current result set.

Returns:

Type	Description
`Iterator[tuple[object, ...]]`	An iterator over the remaining rows.

Examples:

cursor.execute("SELECT id, name FROM users")
for row in cursor:
    print(row)

Source code in picopyn/synchronous/cursor.py

def __iter__(self) -> Iterator[tuple[object, ...]]:
    """
    Iterate over the current result set.

    Returns:
        An iterator over the remaining rows.

    Examples:
        ```python
        cursor.execute("SELECT id, name FROM users")
        for row in cursor:
            print(row)
        ```
    """
    self._ensure_result_set()
    return iter(self._cursor)

`close()`

Close the cursor.

Source code in picopyn/synchronous/cursor.py

def close(self) -> None:
    """
    Close the cursor.
    """
    self._cursor.close()

`execute(operation, parameters=None)`

Prepare and execute a database operation (query or command).

Parameters:

Name	Type	Description	Default
`operation`	`QueryNoTemplate`	database operation to be prepared.	required
`parameters`	`Params \| None`	sequence or mapping that will be bound to variables in the operation.	`None`

Returns:

Type	Description
`Cursor`	This cursor, to allow chaining (e.g. `cursor.execute(query).fetchall()`).

Examples:

DDL example:

ddl = 'CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id);'
cursor.execute(ddl)

DML example:

cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))

Source code in picopyn/synchronous/cursor.py

def execute(self, operation: QueryNoTemplate, parameters: Params | None = None) -> "Cursor":
    """
    Prepare and execute a database operation (query or command).

    Args:
        operation: database operation to be prepared.
        parameters: sequence or mapping that will be bound to variables in the operation.

    Returns:
        This cursor, to allow chaining (e.g. `cursor.execute(query).fetchall()`).

    Examples:
        DDL example:

        ```python
        ddl = 'CREATE TABLE "warehouse" (id INTEGER NOT NULL, item TEXT NOT NULL, PRIMARY KEY (id)) USING memtx DISTRIBUTED BY (id);'
        cursor.execute(ddl)
        ```

        DML example:

        ```python
        cursor.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
        ```
    """
    self._ensure_open()

    try:
        self._cursor.execute(operation, parameters)
    except psycopg.Error as error:
        _raise_dbapi_error(error)

    return self

`executemany(operation, seq_of_parameters)`

Prepare a database operation and execute it against all parameter sequences or mappings found in the sequence seq_of_parameters.

Parameters:

Name	Type	Description	Default
`operation`	`QueryNoTemplate`	database operation to be prepared.	required
`seq_of_parameters`	`Iterable[Params]`	all parameter sequences or mappings that will be bound to variables in the operation.	required

Examples:

cursor.executemany(
    'INSERT INTO "warehouse" VALUES (%s, %s)',
    [(1, "test1"), (2, "test2")],
)

Source code in picopyn/synchronous/cursor.py

def executemany(self, operation: QueryNoTemplate, seq_of_parameters: Iterable[Params]) -> None:
    """
    Prepare a database operation and execute it against all parameter sequences
    or mappings found in the sequence seq_of_parameters.

    Args:
        operation: database operation to be prepared.
        seq_of_parameters: all parameter sequences or mappings that will be bound to variables in the operation.

    Examples:
        ```python
        cursor.executemany(
            'INSERT INTO "warehouse" VALUES (%s, %s)',
            [(1, "test1"), (2, "test2")],
        )
        ```
    """
    self._ensure_open()
    try:
        self._cursor.executemany(operation, seq_of_parameters)
    except psycopg.Error as error:
        _raise_dbapi_error(error)

`fetchall()`

Fetch all remaining rows from the current result set.

Returns:

Type	Description
`list[tuple[object, ...]]`	A list of the remaining rows. Empty if no more rows are available.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
rows = cursor.fetchall()
# [(1, 'test')]

Source code in picopyn/synchronous/cursor.py

def fetchall(self) -> list[tuple[object, ...]]:
    """
    Fetch all remaining rows from the current result set.

    Returns:
        A list of the remaining rows. Empty if no more rows are available.

    Examples:
        ```python
        cursor.execute('SELECT * FROM "warehouse"')
        rows = cursor.fetchall()
        # [(1, 'test')]
        ```
    """
    self._ensure_result_set()
    return self._cursor.fetchall()

`fetchmany(size=None)`

Fetch the next set of rows from the current result set.

Parameters:

Name	Type	Description	Default
`size`	`int \| None`	Maximum number of rows to fetch. If omitted, `arraysize` is used.	`None`

Returns:

Type	Description
`list[tuple[object, ...]]`	A list of the fetched rows. Empty if no more rows are available.

Examples:

Use the previously configured arraysize:

cursor.arraysize = 100
rows = cursor.fetchmany()

Override the batch size for a single call:

rows = cursor.fetchmany(50)

Source code in picopyn/synchronous/cursor.py

def fetchmany(self, size: int | None = None) -> list[tuple[object, ...]]:
    """
    Fetch the next set of rows from the current result set.

    Args:
        size: Maximum number of rows to fetch. If omitted, `arraysize` is used.

    Returns:
        A list of the fetched rows. Empty if no more rows are available.

    Examples:
        Use the previously configured `arraysize`:

        ```python
        cursor.arraysize = 100
        rows = cursor.fetchmany()
        ```

        Override the batch size for a single call:

        ```python
        rows = cursor.fetchmany(50)
        ```
    """
    self._ensure_result_set()

    if size is None:
        return self._cursor.fetchmany()
    return self._cursor.fetchmany(size)

`fetchone()`

Fetch the next row from the current result set.

Returns:

Type	Description
`tuple[object, ...] \| None`	The next row, or None if no more rows are available.

Examples:

cursor.execute('SELECT * FROM "warehouse"')
row = cursor.fetchone()
# (1, 'test')

Source code in picopyn/synchronous/cursor.py

def fetchone(self) -> tuple[object, ...] | None:
    """
    Fetch the next row from the current result set.

    Returns:
        The next row, or None if no more rows are available.

    Examples:
        ```python
        cursor.execute('SELECT * FROM "warehouse"')
        row = cursor.fetchone()
        # (1, 'test')
        ```
    """
    self._ensure_result_set()
    return self._cursor.fetchone()