or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

connection.mdcursor.mderrors.mdindex.mdmeta.mdsqlalchemy.mdtypes.md
tile.json

tessl/pypi-phoenixdb

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/phoenixdb@1.2.x

To install, run

npx @tessl/cli install tessl/pypi-phoenixdb@1.2.0

index.mddocs/

phoenixdb

Python database adapter library that enables developers to connect to and interact with Apache Phoenix databases using the remote query server interface. It implements the standard Python DB API 2.0 specification for database connectivity and includes partial SQLAlchemy dialect support, making it compatible with most Python database applications and ORM frameworks.

Package Information

  • Package Name: phoenixdb
  • Language: Python
  • Installation: pip install phoenixdb

Core Imports

import phoenixdb

Common for working with cursors:

import phoenixdb.cursor
from phoenixdb.cursor import DictCursor

For SQLAlchemy integration:

from phoenixdb.sqlalchemy_phoenix import PhoenixDialect

Basic Usage

import phoenixdb

# Connect to Phoenix query server
database_url = 'http://localhost:8765/'
conn = phoenixdb.connect(database_url, autocommit=True)

# Execute queries
cursor = conn.cursor()
cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, username VARCHAR)")
cursor.execute("UPSERT INTO users VALUES (?, ?)", (1, 'admin'))
cursor.execute("SELECT * FROM users")
results = cursor.fetchall()
print(results)

# Use dictionary cursor for named access
dict_cursor = conn.cursor(cursor_factory=phoenixdb.cursor.DictCursor)
dict_cursor.execute("SELECT * FROM users WHERE id=1")
user = dict_cursor.fetchone()  # Returns dict instead of tuple
print(user['USERNAME'])

# Close resources
cursor.close()
conn.close()

Context manager usage:

import phoenixdb

with phoenixdb.connect('http://localhost:8765/', autocommit=True) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        for row in cursor:  # Cursor supports iteration
            print(row)

SQLAlchemy integration:

from sqlalchemy import create_engine, text

# Create engine for Phoenix
engine = create_engine('phoenix://localhost:8765')

# Execute queries
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users"))
    for row in result:
        print(row)

Architecture

phoenixdb follows the DB API 2.0 specification with these key components:

  • Connection Management: Handles Phoenix query server communication via Apache Avatica protocol
  • Cursor Operations: Provides result set iteration, prepared statements, and bulk operations
  • Type System: Converts between Python and Phoenix/Java types with full precision support
  • Authentication: Supports multiple authentication mechanisms (BASIC, DIGEST, SPNEGO/Kerberos)
  • Error Handling: Complete DB API 2.0 exception hierarchy with Phoenix-specific error details

DB API 2.0 Compliance

# Module-level attributes
apilevel = "2.0"           # DB API version
threadsafety = 1           # Thread safety level
paramstyle = 'qmark'       # Parameter placeholder style (?)

Capabilities

Connection Management

Database connection establishment, configuration, and lifecycle management with support for various authentication mechanisms and connection parameters.

def connect(url, max_retries=None, auth=None, authentication=None,
           avatica_user=None, avatica_password=None, truststore=None,
           verify=None, do_as=None, user=None, password=None,
           extra_headers=None, **kwargs):
    """
    Connects to a Phoenix query server.

    Parameters:
    - url (str): URL to Phoenix query server (e.g., 'http://localhost:8765/')
    - autocommit (bool): Switch connection to autocommit mode
    - readonly (bool): Switch connection to readonly mode
    - max_retries (int): Maximum number of retries for connection errors
    - cursor_factory: Default cursor factory class
    - auth: Authentication configuration object
    - authentication (str): Authentication mechanism ('BASIC', 'DIGEST', 'SPNEGO', 'NONE')
    - avatica_user (str): Username for BASIC/DIGEST authentication
    - avatica_password (str): Password for BASIC/DIGEST authentication
    - truststore (str): Path to PEM file for server certificate verification
    - verify: Certificate verification configuration
    - do_as (str): Username to impersonate (Hadoop doAs)
    - user (str): Alias for avatica_user or do_as depending on authentication
    - password (str): Alias for avatica_password
    - extra_headers (dict): Additional HTTP headers

    Returns:
    Connection: Database connection object
    """

Connection Management

Query Execution

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

class Cursor:
    def execute(self, operation, parameters=None): ...
    def executemany(self, operation, seq_of_parameters): ...
    def fetchone(self): ...
    def fetchmany(self, size=None): ...
    def fetchall(self): ...

Query Execution

Type System

Comprehensive type system for converting between Python and Phoenix data types, including support for date/time, numeric, binary, and array types.

# Type constructor functions
def Date(year, month, day): ...
def Time(hour, minute, second): ...
def Timestamp(year, month, day, hour, minute, second): ...
def Binary(value): ...

# Type constants for comparison
STRING: ColumnType
BINARY: ColumnType
NUMBER: ColumnType
DATETIME: ColumnType
BOOLEAN: ColumnType

Type System

Error Handling

Complete DB API 2.0 exception hierarchy with Phoenix-specific error information including SQL state codes and detailed error messages.

class Error(Exception):
    @property
    def message(self): ...
    @property
    def code(self): ...
    @property
    def sqlstate(self): ...
    @property
    def cause(self): ...

Error Handling

Database Metadata

Access to Phoenix database metadata including catalogs, schemas, tables, columns, primary keys, and indexes through JDBC-compatible interface.

class Meta:
    def get_catalogs(self): ...
    def get_schemas(self, catalog=None, schemaPattern=None): ...
    def get_tables(self, catalog=None, schemaPattern=None, tableNamePattern=None, typeList=None): ...
    def get_columns(self, catalog=None, schemaPattern=None, tableNamePattern=None, columnNamePattern=None): ...

Database Metadata

SQLAlchemy Integration

Partial SQLAlchemy dialect implementation supporting textual SQL execution and basic database operations within SQLAlchemy applications.

class PhoenixDialect(DefaultDialect):
    name = "phoenix"
    driver = "phoenixdb"

SQLAlchemy Integration