tessl/pypi-phoenixdb

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

Overview

Eval results

Files

Type System

Name: tessl/pypi-phoenixdb
Author: tessl

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

Import Statement

from phoenixdb import Date, Time, Timestamp, DateFromTicks, TimeFromTicks, TimestampFromTicks, Binary
from phoenixdb import STRING, BINARY, NUMBER, DATETIME, BOOLEAN, ROWID
from phoenixdb.types import TypeHelper, ColumnType

Common import pattern:

import phoenixdb.types

Capabilities

Type Constructor Functions

DB API 2.0 compliant functions for creating typed values compatible with Phoenix database operations.

def Date(year, month, day):
    """
    Constructs a date object.

    Parameters:
    - year (int): Year component
    - month (int): Month component (1-12)
    - day (int): Day component

    Returns:
    datetime.date: Date object
    """

def Time(hour, minute, second):
    """
    Constructs a time object.

    Parameters:
    - hour (int): Hour component (0-23)
    - minute (int): Minute component (0-59)
    - second (int): Second component (0-59)

    Returns:
    datetime.time: Time object
    """

def Timestamp(year, month, day, hour, minute, second):
    """
    Constructs a datetime/timestamp object.

    Parameters:
    - year (int): Year component
    - month (int): Month component (1-12)
    - day (int): Day component
    - hour (int): Hour component (0-23)
    - minute (int): Minute component (0-59)
    - second (int): Second component (0-59)

    Returns:
    datetime.datetime: Datetime object
    """

def DateFromTicks(ticks):
    """
    Constructs a date object from UNIX timestamp.

    Parameters:
    - ticks (float): UNIX timestamp

    Returns:
    datetime.date: Date object
    """

def TimeFromTicks(ticks):
    """
    Constructs a time object from UNIX timestamp.

    Parameters:
    - ticks (float): UNIX timestamp

    Returns:
    datetime.time: Time object
    """

def TimestampFromTicks(ticks):
    """
    Constructs a datetime object from UNIX timestamp.

    Parameters:
    - ticks (float): UNIX timestamp

    Returns:
    datetime.datetime: Datetime object
    """

def Binary(value):
    """
    Constructs a binary object for handling binary data.

    Parameters:
    - value: Binary data (bytes, bytearray, or convertible)

    Returns:
    bytes: Binary data object
    """

Type Constants

Column type objects for type comparison and identification in cursor descriptions.

STRING: ColumnType
"""Type object for string-based columns (VARCHAR, CHAR)."""

BINARY: ColumnType
"""Type object for binary columns (BINARY, VARBINARY)."""

NUMBER: ColumnType
"""
Type object for numeric columns including:
INTEGER, UNSIGNED_INT, BIGINT, UNSIGNED_LONG, TINYINT, UNSIGNED_TINYINT,
SMALLINT, UNSIGNED_SMALLINT, FLOAT, UNSIGNED_FLOAT, DOUBLE, UNSIGNED_DOUBLE, DECIMAL
"""

DATETIME: ColumnType
"""
Type object for date/time columns including:
TIME, DATE, TIMESTAMP, UNSIGNED_TIME, UNSIGNED_DATE, UNSIGNED_TIMESTAMP
"""

BOOLEAN: ColumnType
"""Type object for boolean columns (Phoenix-specific extension)."""

ROWID: ColumnType
"""Type object for row identifiers (DB API 2.0 compatibility only)."""

Type Helper Class

Utility class for internal type mapping and conversion between JDBC types and Python types.

class TypeHelper:
    """Static utility methods for type conversion and mapping."""

    @staticmethod
    def from_param(param):
        """
        Retrieves type information from AvaticaParameter object.

        Parameters:
        - param: Protobuf AvaticaParameter object

        Returns:
        tuple: (field_name, rep, mutate_to, cast_from, is_array)
               - field_name: Attribute in TypedValue
               - rep: common_pb2.Rep enum
               - mutate_to: Function to cast to Phoenix values
               - cast_from: Function to cast from Phoenix values
               - is_array: Boolean indicating array parameter

        Raises:
        NotImplementedError: For unsupported JDBC type codes
        """

    @staticmethod
    def from_column(column):
        """
        Retrieves type information from column metadata.

        Parameters:
        - column: Protobuf TypedValue object

        Returns:
        tuple: (field_name, rep, mutate_to, cast_from)
               - field_name: Attribute in TypedValue
               - rep: common_pb2.Rep enum
               - mutate_to: Function to cast to Phoenix values
               - cast_from: Function to cast from Phoenix values

        Raises:
        NotImplementedError: For unsupported JDBC type codes
        """

Column Type Class

Type comparison objects for cursor description analysis.

class ColumnType:
    """Type object for column type comparison."""

    def __init__(self, eq_types):
        """
        Initialize with list of equivalent type names.

        Parameters:
        - eq_types (list): List of type names this object represents
        """

    def __eq__(self, other):
        """
        Compare with type name.

        Parameters:
        - other (str): Type name to compare

        Returns:
        bool: True if other matches this type
        """

Usage Examples

Creating Typed Values

import phoenixdb
from datetime import datetime, date, time

# Date values
birth_date = phoenixdb.Date(1990, 5, 15)
today = phoenixdb.DateFromTicks(time.time())

# Time values
lunch_time = phoenixdb.Time(12, 30, 0)
current_time = phoenixdb.TimeFromTicks(time.time())

# Timestamp values
event_time = phoenixdb.Timestamp(2023, 12, 25, 14, 30, 0)
now = phoenixdb.TimestampFromTicks(time.time())

# Binary data
binary_data = phoenixdb.Binary(b'\x00\x01\x02\x03')

# Use in queries
cursor.execute("INSERT INTO events (id, event_date, event_time, data) VALUES (?, ?, ?, ?)",
               (1, birth_date, lunch_time, binary_data))

Type Checking in Results

cursor.execute("SELECT * FROM mixed_table")

# Check column types
for col in cursor.description:
    if col.type_code == phoenixdb.STRING:
        print(f"{col.name} is a string column")
    elif col.type_code == phoenixdb.NUMBER:
        print(f"{col.name} is a numeric column")
    elif col.type_code == phoenixdb.DATETIME:
        print(f"{col.name} is a date/time column")
    elif col.type_code == phoenixdb.BOOLEAN:
        print(f"{col.name} is a boolean column")

Working with Python Native Types

from datetime import datetime, date
from decimal import Decimal

# phoenixdb automatically handles Python native types
cursor = conn.cursor()

# Python datetime objects work directly
current_datetime = datetime.now()
cursor.execute("INSERT INTO logs (timestamp, message) VALUES (?, ?)",
               (current_datetime, "Log message"))

# Python date objects
event_date = date.today()
cursor.execute("INSERT INTO events (event_date, description) VALUES (?, ?)",
               (event_date, "Event description"))

# Decimal for precise numeric values
price = Decimal('199.95')
cursor.execute("INSERT INTO products (name, price) VALUES (?, ?)",
               ("Product Name", price))

Phoenix-Specific Type Handling

Array Types

Phoenix supports array columns, which phoenixdb handles automatically.

# Insert array values
cursor.execute("CREATE TABLE test_arrays (id INTEGER, numbers INTEGER ARRAY)")

# Arrays as Python lists
numbers = [1, 2, 3, 4, 5]
cursor.execute("INSERT INTO test_arrays VALUES (?, ?)", (1, numbers))

# Query array columns
cursor.execute("SELECT * FROM test_arrays")
row = cursor.fetchone()
array_values = row[1]  # Returns Python list
print(f"Array values: {array_values}")

Precision Handling

Phoenix timestamps have nanosecond precision, but the remote protocol truncates to milliseconds.

from datetime import datetime

# Microsecond precision is preserved up to millisecond level
precise_time = datetime(2023, 12, 25, 14, 30, 45, 123456)  # 123.456 ms
cursor.execute("INSERT INTO events (timestamp) VALUES (?)", (precise_time,))

# Retrieved value will have millisecond precision
cursor.execute("SELECT timestamp FROM events")
retrieved_time = cursor.fetchone()[0]
# Retrieved: 2023-12-25 14:30:45.123000 (microseconds truncated to ms)

Unsigned Types

Phoenix supports unsigned numeric types that map to appropriate Python types.

# Phoenix unsigned types automatically handled
cursor.execute("CREATE TABLE test_unsigned (id UNSIGNED_INT, big_num UNSIGNED_LONG)")

# Python int values work directly
cursor.execute("INSERT INTO test_unsigned VALUES (?, ?)", (4294967295, 18446744073709551615))

cursor.execute("SELECT * FROM test_unsigned")
row = cursor.fetchone()
print(f"Unsigned int: {row[0]}, Unsigned long: {row[1]}")

Binary Data

# Insert binary data
image_data = open('image.png', 'rb').read()
binary_obj = phoenixdb.Binary(image_data)

cursor.execute("CREATE TABLE images (id INTEGER, data VARBINARY)")
cursor.execute("INSERT INTO images VALUES (?, ?)", (1, binary_obj))

# Retrieve binary data
cursor.execute("SELECT data FROM images WHERE id = ?", (1,))
retrieved_data = cursor.fetchone()[0]  # Returns bytes object

Internal Type Conversion

Phoenix uses Java/JDBC type system internally. phoenixdb handles conversion automatically:

Java to Python Mapping

Java SQL Date → Python datetime.date
Java SQL Time → Python datetime.time
Java SQL Timestamp → Python datetime.datetime
Java BigDecimal → Python Decimal (for precise numeric values)
Java byte[] → Python bytes
Java primitive types → Python int/float/bool

Conversion Functions

Internal utility functions handle the conversion:

# These functions are used internally but available if needed
from phoenixdb.types import (
    date_from_java_sql_date, date_to_java_sql_date,
    time_from_java_sql_time, time_to_java_sql_time,
    datetime_from_java_sql_timestamp, datetime_to_java_sql_timestamp
)

# Manual conversion (rarely needed)
java_date_ms = date_to_java_sql_date(date.today())
python_date = date_from_java_sql_date(java_date_ms)

Java Conversion Function Definitions

Complete API definitions for the Java conversion functions.

def date_from_java_sql_date(n):
    """
    Converts Java SQL Date (milliseconds since epoch) to Python date.

    Parameters:
    - n (int): Milliseconds since epoch (Java Date format)

    Returns:
    datetime.date: Python date object
    """

def date_to_java_sql_date(d):
    """
    Converts Python date to Java SQL Date format.

    Parameters:
    - d (datetime.date): Python date object

    Returns:
    int: Milliseconds since epoch (Java Date format)
    """

def time_from_java_sql_time(n):
    """
    Converts Java SQL Time (milliseconds since midnight) to Python time.

    Parameters:
    - n (int): Milliseconds since midnight (Java Time format)

    Returns:
    datetime.time: Python time object
    """

def time_to_java_sql_time(t):
    """
    Converts Python time to Java SQL Time format.

    Parameters:
    - t (datetime.time): Python time object

    Returns:
    int: Milliseconds since midnight (Java Time format)
    """

def datetime_from_java_sql_timestamp(n):
    """
    Converts Java SQL Timestamp (milliseconds since epoch) to Python datetime.

    Parameters:
    - n (int): Milliseconds since epoch (Java Timestamp format)

    Returns:
    datetime.datetime: Python datetime object
    """

def datetime_to_java_sql_timestamp(d):
    """
    Converts Python datetime to Java SQL Timestamp format.

    Parameters:
    - d (datetime.datetime): Python datetime object

    Returns:
    int: Milliseconds since epoch (Java Timestamp format)
    """

Install with Tessl CLI