tessl/pypi-phoenixdb

Python database adapter library for Apache Phoenix databases implementing DB API 2.0 and partial SQLAlchemy support

Overview

Eval results

Files

Query Execution

Name: tessl/pypi-phoenixdb
Author: tessl

Cursor-based query execution with support for prepared statements, parameter binding, result set iteration, and bulk operations.

Import Statement

import phoenixdb.cursor
from phoenixdb.cursor import Cursor, DictCursor, ColumnDescription

Capabilities

Cursor Creation

Cursors are created from connections and provide the interface for executing SQL statements and retrieving results.

class Connection:
    def cursor(self, cursor_factory=None):
        """
        Creates a new cursor.

        Parameters:
        - cursor_factory: Cursor class to use (default: Cursor)
                         Common options: Cursor, DictCursor

        Returns:
        Cursor: New cursor object

        Raises:
        ProgrammingError: If connection is closed
        """

Standard Cursor

The standard cursor returns results as tuples and provides full DB API 2.0 functionality.

class Cursor:
    """Database cursor for executing queries and iterating over results."""

    def __init__(self, connection, id=None): ...

    def close(self):
        """
        Closes the cursor. No further operations allowed after closing.
        Automatically called when using with statement.
        """

    def execute(self, operation, parameters=None):
        """
        Executes SQL statement with optional parameters.

        Parameters:
        - operation (str): SQL statement to execute
        - parameters (list/tuple): Parameter values for placeholders

        Raises:
        ProgrammingError: SQL syntax errors, parameter count mismatch
        DatabaseError: Database execution errors
        """

    def executemany(self, operation, seq_of_parameters):
        """
        Executes SQL statement with multiple parameter sets (bulk operation).

        Parameters:
        - operation (str): SQL statement to execute
        - seq_of_parameters (list): List of parameter tuples/lists

        Returns:
        List of update counts

        Raises:
        ProgrammingError: SQL syntax errors, parameter issues
        DatabaseError: Database execution errors
        """

    def fetchone(self):
        """
        Fetches next result row.

        Returns:
        tuple: Next row or None if no more rows

        Raises:
        ProgrammingError: No select statement executed
        """

    def fetchmany(self, size=None):
        """
        Fetches multiple result rows.

        Parameters:
        - size (int): Number of rows to fetch (default: arraysize)

        Returns:
        list: List of row tuples
        """

    def fetchall(self):
        """
        Fetches all remaining result rows.

        Returns:
        list: List of all remaining row tuples
        """

    def setinputsizes(self, sizes):
        """DB API 2.0 compatibility method (no-op)."""

    def setoutputsize(self, size, column=None):
        """DB API 2.0 compatibility method (no-op)."""

Column Description

Result set column metadata structure returned by cursor.description.

ColumnDescription = collections.namedtuple('ColumnDescription', 'name type_code display_size internal_size precision scale null_ok')
"""
Named tuple for representing column metadata in cursor.description.

Fields:
- name (str): Column name
- type_code (ColumnType): Type code object for comparison
- display_size (int): Display size (may be None)
- internal_size (int): Internal size (may be None)
- precision (int): Numeric precision (may be None)
- scale (int): Numeric scale (may be None)
- null_ok (bool/None): Whether column allows NULL values
"""

Cursor Properties

class Cursor:
    @property
    def description(self):
        """
        Column descriptions as ColumnDescription namedtuples.
        None if no result set available.

        Returns:
        list: List of ColumnDescription tuples with fields:
              - name: Column name
              - type_code: Type code
              - display_size: Display size
              - internal_size: Internal size
              - precision: Numeric precision
              - scale: Numeric scale
              - null_ok: Nullable flag (True/False/None)
        """

    @property
    def rowcount(self):
        """
        Number of rows affected by last DML statement.
        -1 for SELECT statements or unknown count.

        Returns:
        int: Row count or -1
        """

    @property
    def rownumber(self):
        """
        Current 0-based row index in result set.
        None if position cannot be determined.

        Returns:
        int: Current row index or None
        """

    @property
    def connection(self):
        """
        Associated Connection object.

        Returns:
        Connection: Connection that created this cursor
        """

    @property
    def closed(self):
        """
        Boolean indicating if cursor is closed.

        Returns:
        bool: True if closed, False if open
        """

    arraysize = 1
    """Number of rows fetchmany() retrieves by default."""

    itersize = 2000
    """Number of rows to fetch per network round trip during iteration."""

Dictionary Cursor

Returns results as dictionaries with column names as keys instead of tuples.

class DictCursor(Cursor):
    """
    Cursor that returns results as dictionaries with column names as keys.

    Inherits all methods from Cursor but fetchone(), fetchmany(), fetchall()
    and iteration return dictionaries instead of tuples.
    """

    def fetchone(self):
        """
        Fetches next result row as dictionary.

        Returns:
        dict: Row with column names as keys, or None if no more rows
        """

    def fetchmany(self, size=None):
        """
        Fetches multiple result rows as list of dictionaries.

        Parameters:
        - size (int): Number of rows to fetch

        Returns:
        list: List of dictionaries with column names as keys
        """

    def fetchall(self):
        """
        Fetches all remaining result rows as list of dictionaries.

        Returns:
        list: List of dictionaries with column names as keys
        """

Usage example:

cursor = conn.cursor(cursor_factory=phoenixdb.cursor.DictCursor)
cursor.execute("SELECT id, username FROM users WHERE id=1")
user = cursor.fetchone()  # Returns {'id': 1, 'username': 'admin'}
print(user['username'])

Context Manager Support

Cursors support context manager protocol for automatic cleanup.

class Cursor:
    def __enter__(self):
        """Context manager entry."""
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """Context manager exit with automatic cursor closing."""

Iterator Protocol

Cursors can be used as iterators to process result sets row by row.

class Cursor:
    def __iter__(self):
        """Returns self as iterator."""
        return self

    def __next__(self):
        """
        Fetches next row for iteration.

        Returns:
        tuple: Next row

        Raises:
        StopIteration: When no more rows available
        """

    # Python 2 compatibility
    next = __next__

Usage Examples

Basic Query Execution

cursor = conn.cursor()

# Simple SELECT
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# Parameterized query
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
user = cursor.fetchone()

# INSERT with parameters
cursor.execute("INSERT INTO users (id, username) VALUES (?, ?)", (2, 'john'))

Bulk Operations

cursor = conn.cursor()

# Bulk INSERT
users = [(1, 'admin'), (2, 'john'), (3, 'jane')]
cursor.executemany("INSERT INTO users (id, username) VALUES (?, ?)", users)

# Check affected row counts
print(f"Inserted {cursor.rowcount} rows")

Result Set Iteration

cursor = conn.cursor()
cursor.execute("SELECT * FROM large_table")

# Iterate over results without loading all into memory
for row in cursor:
    process_row(row)
    # Automatically fetches more rows as needed

Fetch Size Control

cursor = conn.cursor()
cursor.arraysize = 100  # fetchmany() default size
cursor.itersize = 5000  # Network fetch size

cursor.execute("SELECT * FROM users")

# Fetch in chunks
while True:
    rows = cursor.fetchmany()
    if not rows:
        break
    process_rows(rows)

Column Information

cursor = conn.cursor()
cursor.execute("SELECT id, username, created_date FROM users")

# Examine column metadata
for col in cursor.description:
    print(f"Column: {col.name}, Type: {col.type_code}, Nullable: {col.null_ok}")

# Access results
row = cursor.fetchone()
print(f"ID: {row[0]}, Username: {row[1]}, Created: {row[2]}")

Phoenix-Specific Operations

cursor = conn.cursor()

# Phoenix UPSERT operation
cursor.execute("UPSERT INTO users VALUES (?, ?)", (1, 'updated_admin'))

# Array parameter support
cursor.execute("SELECT * FROM table WHERE id = ANY(?)", ([1, 2, 3],))

# Phoenix timestamp precision handling
from datetime import datetime
timestamp = datetime.now()
cursor.execute("INSERT INTO events (id, timestamp) VALUES (?, ?)", (1, timestamp))

Advanced Cursor Operations

Prepared Statement Reuse

cursor = conn.cursor()

# Prepare statement once
cursor.execute("INSERT INTO users (id, username) VALUES (?, ?)", (1, 'first'))

# Reuse prepared statement with different parameters
cursor.execute("INSERT INTO users (id, username) VALUES (?, ?)", (2, 'second'))
cursor.execute("INSERT INTO users (id, username) VALUES (?, ?)", (3, 'third'))

Error Handling

import phoenixdb

cursor = conn.cursor()
try:
    cursor.execute("SELECT * FROM nonexistent_table")
except phoenixdb.ProgrammingError as e:
    print(f"SQL Error: {e.message}")
    print(f"SQL State: {e.sqlstate}")
    print(f"Error Code: {e.code}")
except phoenixdb.OperationalError as e:
    print(f"Database Error: {e.message}")

Install with Tessl CLI