CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-vertica-python

Official native Python client for the Vertica database.

Overview
Eval results
Files

data-types.mddocs/

Data Types

Comprehensive data type support including DB-API 2.0 type constructors, Vertica-specific types, and customizable type conversion system. Handles complex types like arrays, sets, and custom data structures with utilities for type inspection and conversion.

Capabilities

DB-API 2.0 Type Constructors

Standard type constructors for creating database-compatible objects.

def Date(year: int, month: int, day: int):
    """
    Construct a date object.
    
    Parameters:
    - year (int): Year value
    - month (int): Month value (1-12)
    - day (int): Day value (1-31)
    
    Returns:
    datetime.date: Date object
    """

def Time(hour: int, minute: int, second: int):
    """
    Construct a time object.
    
    Parameters:
    - hour (int): Hour value (0-23)
    - minute (int): Minute value (0-59)
    - second (int): Second value (0-59)
    
    Returns:
    datetime.time: Time object
    """

def Timestamp(year: int, month: int, day: int, hour: int, minute: int, second: int):
    """
    Construct a timestamp object.
    
    Parameters:
    - year (int): Year value
    - month (int): Month value (1-12)
    - day (int): Day value (1-31)
    - hour (int): Hour value (0-23)
    - minute (int): Minute value (0-59)
    - second (int): Second value (0-59)
    
    Returns:
    datetime.datetime: Timestamp object
    """

def DateFromTicks(ticks: float):
    """
    Construct a date from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp (seconds since epoch)
    
    Returns:
    datetime.date: Date object
    """

def TimeFromTicks(ticks: float):
    """
    Construct a time from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp (seconds since epoch)
    
    Returns:
    datetime.time: Time object
    """

def TimestampFromTicks(ticks: float):
    """
    Construct a timestamp from Unix timestamp.
    
    Parameters:
    - ticks (float): Unix timestamp (seconds since epoch)
    
    Returns:
    datetime.time: Time object (Note: This appears to be a bug in the source code - should return datetime)
    """

def Binary(string):
    """
    Construct a binary object for handling binary data.
    
    Parameters:
    - string: Binary string data
    
    Returns:
    Bytea: Binary object wrapper
    """

DB-API 2.0 Type Objects

Type objects for column type comparison and identification.

STRING = VerticaType([VerticaType.CHAR, VerticaType.VARCHAR, VerticaType.BINARY,
                     VerticaType.VARBINARY, VerticaType.UNKNOWN,
                     VerticaType.LONGVARBINARY, VerticaType.LONGVARCHAR])
"""Type object for string-based column types."""

BINARY = VerticaType([VerticaType.BINARY, VerticaType.VARBINARY, VerticaType.LONGVARBINARY])
"""Type object for binary column types."""

NUMBER = VerticaType([VerticaType.INT8, VerticaType.FLOAT8, VerticaType.NUMERIC])
"""Type object for numeric column types."""

DATETIME = VerticaType([VerticaType.DATE, VerticaType.TIME, VerticaType.TIMETZ,
                       VerticaType.TIMESTAMP, VerticaType.TIMESTAMPTZ,
                       VerticaType.INTERVAL, VerticaType.INTERVALYM])
"""Type object for date/time column types."""

ROWID = VerticaType([])
"""Type object for row ID types (empty for Vertica)."""

VerticaType Class

Type constants and comparison functionality for all Vertica data types.

class VerticaType:
    """
    Type constants and comparison class for Vertica data types.
    Contains all Vertica type OID constants and comparison methods.
    """
    
    # Basic Types
    UNKNOWN = 4
    BOOL = 5
    INT8 = 6
    FLOAT8 = 7
    CHAR = 8
    VARCHAR = 9
    DATE = 10
    TIME = 11
    TIMESTAMP = 12
    TIMESTAMPTZ = 13
    INTERVAL = 14
    INTERVALYM = 114
    TIMETZ = 15
    NUMERIC = 16
    VARBINARY = 17
    UUID = 20
    LONGVARCHAR = 115
    LONGVARBINARY = 116
    BINARY = 117
    
    # Complex Types
    ROW = 300
    ARRAY = 301
    MAP = 302
    
    # One-dimensional arrays (selected examples)
    ARRAY1D_BOOL = 1505
    ARRAY1D_INT8 = 1506
    ARRAY1D_FLOAT8 = 1507
    # ... (many more array type constants)
    
    # One-dimensional sets (selected examples)
    SET_BOOL = 2705
    SET_INT8 = 2706
    # ... (many more set type constants)
    
    def __init__(self, type_list=None):
        """
        Initialize VerticaType with list of type OIDs.
        
        Parameters:
        - type_list (list, optional): List of type OIDs for comparison
        """
    
    def __eq__(self, other) -> bool:
        """
        Compare type objects for equality.
        
        Parameters:
        - other: Type OID or VerticaType to compare
        
        Returns:
        bool: True if types match, False otherwise
        """
    
    def __ne__(self, other) -> bool:
        """
        Compare type objects for inequality.
        
        Parameters:
        - other: Type OID or VerticaType to compare
        
        Returns:
        bool: True if types don't match, False otherwise
        """

Bytea Class

Binary string class for handling binary data with proper encoding.

class Bytea:
    """
    Binary string class for handling binary data.
    """
    
    def __init__(self, data):
        """
        Initialize Bytea object with binary data.
        
        Parameters:
        - data: Binary data (bytes, str, or other binary-compatible type)
        """
    
    def __str__(self) -> str:
        """
        Get string representation of binary data.
        
        Returns:
        str: String representation
        """
    
    def __repr__(self) -> str:
        """
        Get detailed string representation.
        
        Returns:
        str: Detailed representation
        """
    
    def __eq__(self, other) -> bool:
        """
        Compare Bytea objects for equality.
        
        Parameters:
        - other: Other Bytea object to compare
        
        Returns:
        bool: True if equal, False otherwise
        """

Type Inspection Utilities

Utility functions for inspecting and working with Vertica data types.

def getTypeName(data_type_oid: int, type_modifier: int = -1) -> str:
    """
    Get human-readable type name from type OID and modifier.
    
    Parameters:
    - data_type_oid (int): Vertica type OID
    - type_modifier (int): Type modifier for precision/scale info (default: -1)
    
    Returns:
    str: Human-readable type name (e.g., 'VARCHAR(100)', 'NUMERIC(10,2)')
    """

def getComplexElementType(data_type_oid: int) -> int:
    """
    Get element type OID for complex types (arrays, sets).
    
    Parameters:
    - data_type_oid (int): Complex type OID
    
    Returns:
    int: Element type OID, or 0 if not a complex type
    """

def getIntervalRange(data_type_oid: int, type_modifier: int) -> str:
    """
    Get interval range string for interval types.
    
    Parameters:
    - data_type_oid (int): Interval type OID
    - type_modifier (int): Type modifier containing range information
    
    Returns:
    str: Interval range string (e.g., 'DAY TO SECOND', 'YEAR TO MONTH')
    """

def getIntervalLeadingPrecision(data_type_oid: int, type_modifier: int) -> int:
    """
    Get leading precision of interval types.
    
    Parameters:
    - data_type_oid (int): Interval type OID
    - type_modifier (int): Type modifier containing precision information
    
    Returns:
    int: Leading precision value
    """

def getPrecision(data_type_oid: int, type_modifier: int) -> int:
    """
    Get precision of numeric types.
    
    Parameters:
    - data_type_oid (int): Numeric type OID
    - type_modifier (int): Type modifier containing precision information
    
    Returns:
    int: Precision value, or -1 if not applicable
    """

def getScale(data_type_oid: int, type_modifier: int) -> int:
    """
    Get scale of numeric types.
    
    Parameters:
    - data_type_oid (int): Numeric type OID
    - type_modifier (int): Type modifier containing scale information
    
    Returns:
    int: Scale value, or -1 if not applicable
    """

def getDisplaySize(data_type_oid: int, type_modifier: int) -> int:
    """
    Get display size for a type.
    
    Parameters:
    - data_type_oid (int): Type OID
    - type_modifier (int): Type modifier
    
    Returns:
    int: Display size in characters, or -1 if not applicable
    """

Usage Examples

Basic Type Construction

from vertica_python.datatypes import Date, Time, Timestamp, Binary

# Create date/time objects
birthday = Date(1990, 5, 15)
meeting_time = Time(14, 30, 0)
event_timestamp = Timestamp(2024, 12, 25, 18, 0, 0)

# Create from Unix timestamps
import time
from vertica_python.datatypes import DateFromTicks, TimeFromTicks, TimestampFromTicks
now_date = DateFromTicks(time.time())
now_time = TimeFromTicks(time.time())
now_timestamp = TimestampFromTicks(time.time())

# Handle binary data
image_data = b'\x89PNG\r\n\x1a\n...'  # Binary image data
binary_obj = Binary(image_data)

# Use in queries
with conn.cursor() as cursor:
    cursor.execute(
        "INSERT INTO events (name, date, time, timestamp, image) VALUES (:name, :date, :time, :ts, :img)",
        {
            'name': 'Holiday Party',
            'date': birthday,
            'time': meeting_time,
            'ts': event_timestamp,
            'img': binary_obj
        }
    )

Type Comparison and Detection

from vertica_python.datatypes import getTypeName, STRING, NUMBER, DATETIME, BINARY, VerticaType, getPrecision, getScale

with conn.cursor() as cursor:
    cursor.execute("SELECT name, age, salary, hire_date FROM employees LIMIT 1")
    
    # Examine column types
    for i, col_desc in enumerate(cursor.description):
        name, type_code, display_size, internal_size, precision, scale, null_ok = col_desc
        
        print(f"Column {name}:")
        print(f"  Type code: {type_code}")
        print(f"  Type name: {getTypeName(type_code, -1)}")
        
        # Check type categories
        if type_code in STRING:
            print(f"  Category: String (max length: {display_size})")
        elif type_code in NUMBER:
            print(f"  Category: Number (precision: {precision}, scale: {scale})")
        elif type_code in DATETIME:
            print(f"  Category: Date/Time")
        elif type_code in BINARY:
            print(f"  Category: Binary")
        
        # Get detailed type information
        if type_code == VerticaType.NUMERIC:
            print(f"  Precision: {getPrecision(type_code, precision)}")
            print(f"  Scale: {getScale(type_code, scale)}")

Working with Complex Types

from vertica_python.datatypes import VerticaType, getComplexElementType, getTypeName

# Arrays
with conn.cursor() as cursor:
    cursor.execute("SELECT tags FROM articles WHERE id = 1")
    row = cursor.fetchone()
    tags = row[0]  # This might be a list like ['python', 'database', 'vertica']
    
    # Insert array data
    cursor.execute(
        "INSERT INTO articles (title, tags) VALUES (:title, :tags)",
        {
            'title': 'Database Guide',
            'tags': ['sql', 'database', 'tutorial']
        }
    )

# Check if column is a complex type
with conn.cursor() as cursor:
    cursor.execute("SELECT * FROM articles LIMIT 1")
    for col_desc in cursor.description:
        name, type_code = col_desc[0], col_desc[1]
        
        if type_code == VerticaType.ARRAY:
            element_type = getComplexElementType(type_code)
            element_name = getTypeName(element_type, -1)
            print(f"Column {name} is an array of {element_name}")
        elif getComplexElementType(type_code) is not None:
            element_type = getComplexElementType(type_code)
            element_name = getTypeName(element_type, -1)
            print(f"Column {name} is a set of {element_name}")

Custom Type Conversion

import decimal
from datetime import datetime, timezone
from vertica_python.datatypes import VerticaType

# Custom converter for high-precision decimals
def high_precision_decimal_converter(value):
    """Convert string to high-precision decimal."""
    if value is None:
        return None
    return decimal.Decimal(value)

# Custom converter for timezone-aware timestamps
def timezone_timestamp_converter(value):
    """Convert timestamp to timezone-aware datetime."""
    if value is None:
        return None
    # Assume UTC if no timezone info
    if isinstance(value, datetime) and value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value

with conn.cursor() as cursor:
    # Register custom converters
    cursor.register_sqldata_converter(
        VerticaType.NUMERIC,
        high_precision_decimal_converter
    )
    cursor.register_sqldata_converter(
        VerticaType.TIMESTAMPTZ,
        timezone_timestamp_converter
    )
    
    # Use custom conversion
    cursor.execute("SELECT price, created_at FROM products WHERE id = 1")
    price, created_at = cursor.fetchone()
    
    # price is now a decimal.Decimal with full precision
    # created_at is now timezone-aware
    
    print(f"Price: {price} (type: {type(price)})")
    print(f"Created: {created_at} (timezone: {created_at.tzinfo})")

SQL Literal Adapters

import uuid
import json

# Custom adapter for UUID objects
def uuid_adapter(obj):
    """Convert UUID to SQL literal."""
    return f"'{str(obj)}'"

# Custom adapter for JSON objects
def json_adapter(obj):
    """Convert dict/list to JSON SQL literal."""
    return f"'{json.dumps(obj)}'"

with conn.cursor() as cursor:
    # Register custom adapters
    cursor.register_sql_literal_adapter(uuid.UUID, uuid_adapter)
    cursor.register_sql_literal_adapter(dict, json_adapter)
    cursor.register_sql_literal_adapter(list, json_adapter)
    
    # Use custom adapters
    product_id = uuid.uuid4()
    metadata = {'category': 'electronics', 'tags': ['phone', 'mobile']}
    
    cursor.execute(
        "INSERT INTO products (id, name, metadata) VALUES (:id, :name, :metadata)",
        {
            'id': product_id,  # Automatically converted to SQL string
            'name': 'Smartphone',
            'metadata': metadata  # Automatically converted to JSON string
        }
    )

Working with Intervals

from datetime import timedelta
from vertica_python.datatypes import VerticaType, getIntervalRange, getIntervalLeadingPrecision

with conn.cursor() as cursor:
    # Query with interval type
    cursor.execute("SELECT name, duration FROM events WHERE duration > INTERVAL '1 hour'")
    
    for row in cursor.fetchall():
        name, duration = row
        print(f"Event: {name}")
        
        # Check column metadata for interval details
        for col_desc in cursor.description:
            col_name, type_code, _, _, precision, scale, _ = col_desc
            if col_name == 'duration' and type_code == VerticaType.INTERVAL:
                range_str = getIntervalRange(type_code, precision)
                leading_precision = getIntervalLeadingPrecision(type_code, precision)
                print(f"  Duration: {duration} (range: {range_str}, precision: {leading_precision})")

# Insert interval data
with conn.cursor() as cursor:
    # Vertica accepts various interval formats
    cursor.execute(
        "INSERT INTO events (name, duration) VALUES (:name, :duration)",
        {
            'name': 'Meeting',
            'duration': '2 hours 30 minutes'  # String format
        }
    )
    
    # Or using Python timedelta (if converter is set up)
    duration = timedelta(hours=2, minutes=30)
    cursor.execute(
        "INSERT INTO events (name, duration) VALUES (:name, INTERVAL :duration)",
        {
            'name': 'Workshop',
            'duration': f'{duration.total_seconds()} seconds'
        }
    )

Type Utility Functions

from vertica_python.datatypes import (
    getTypeName, NUMBER, STRING, VerticaType, getPrecision, getScale,
    getDisplaySize, getIntervalRange, getIntervalLeadingPrecision,
    getComplexElementType
)

def analyze_table_schema(cursor, table_name):
    """Analyze and display detailed schema information."""
    cursor.execute(f"SELECT * FROM {table_name} LIMIT 0")  # Get just metadata
    
    print(f"Schema for table: {table_name}")
    print("-" * 60)
    
    for col_desc in cursor.description:
        name, type_code, display_size, internal_size, precision, scale, null_ok = col_desc
        
        # Get type name with modifiers
        type_name = getTypeName(type_code, precision)
        
        print(f"Column: {name}")
        print(f"  Type: {type_name} (OID: {type_code})")
        print(f"  Nullable: {'Yes' if null_ok else 'No'}")
        
        # Type-specific information
        if type_code in NUMBER:
            if getPrecision(type_code, precision) is not None:
                prec = getPrecision(type_code, precision)
                scl = getScale(type_code, scale)
                print(f"  Precision: {prec}, Scale: {scl}")
        
        elif type_code in STRING:
            disp_size = getDisplaySize(type_code, display_size)
            if disp_size is not None:
                print(f"  Max Length: {disp_size}")
        
        elif type_code == VerticaType.INTERVAL:
            range_str = getIntervalRange(type_code, precision)
            leading_prec = getIntervalLeadingPrecision(type_code, precision)
            print(f"  Range: {range_str}")
            print(f"  Leading Precision: {leading_prec}")
        
        elif type_code == VerticaType.ARRAY or getComplexElementType(type_code) is not None:
            element_type = getComplexElementType(type_code)
            if element_type is not None:
                element_name = getTypeName(element_type, -1)
                type_desc = "Array" if type_code == VerticaType.ARRAY else "Set"
                print(f"  {type_desc} of: {element_name}")
        
        print()

# Usage
with conn.cursor() as cursor:
    analyze_table_schema(cursor, 'employees')

Install with Tessl CLI

npx tessl i tessl/pypi-vertica-python

docs

connection-management.md

data-types.md

exception-handling.md

index.md

query-execution.md

tile.json