tessl/pypi-cassandra-driver

Python driver for Apache Cassandra with comprehensive CQL support, connection pooling, and ORM capabilities

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

CQL Types & Data

Name: tessl/pypi-cassandra-driver
Author: tessl

Comprehensive type system with encoding/decoding support for all Cassandra data types, collections, user-defined types, and utility functions. The type system handles conversion between Python objects and Cassandra's native binary format.

Capabilities

Base Type System

Foundation classes for the CQL type system with encoding and decoding capabilities.

class _CassandraType:
    """Abstract base class for all CQL types."""
    
    def serialize(self, value, protocol_version):
        """
        Serialize a Python value to bytes for Cassandra.
        
        Parameters:
        - value: Python value to serialize
        - protocol_version (int): Protocol version to use
        
        Returns:
        bytes: Serialized value
        """
    
    def deserialize(self, byts, protocol_version):
        """
        Deserialize bytes from Cassandra to a Python value.
        
        Parameters:
        - byts (bytes): Serialized data from Cassandra
        - protocol_version (int): Protocol version used
        
        Returns:
        Python value corresponding to the CQL type
        """

class _UnrecognizedType(_CassandraType):
    """Represents an unrecognized CQL type."""
    
    def __init__(self, name):
        """
        Parameters:
        - name (str): Name of the unrecognized type
        """

class _ParameterizedType(_CassandraType):
    """Base class for parameterized types like collections."""
    
    def __init__(self, subtypes, names=None):
        """
        Parameters:
        - subtypes (list): List of parameter types
        - names (list): Parameter names (for user-defined types)
        """

class EmptyValue:
    """Represents empty/null values in CQL."""
    pass

Primitive Types

Basic CQL data types for fundamental values.

class BytesType(_CassandraType):
    """CQL blob type for binary data."""

class DecimalType(_CassandraType):
    """CQL decimal type for high-precision decimal numbers."""

class UUIDType(_CassandraType):
    """CQL uuid type for UUID values."""

class BooleanType(_CassandraType):
    """CQL boolean type."""

class ByteType(_CassandraType):
    """CQL tinyint type (8-bit signed integer)."""

class ShortType(_CassandraType):
    """CQL smallint type (16-bit signed integer)."""

class Int32Type(_CassandraType):
    """CQL int type (32-bit signed integer)."""

class LongType(_CassandraType):
    """CQL bigint type (64-bit signed integer)."""

class IntegerType(_CassandraType):
    """CQL varint type (variable-precision integer)."""

class FloatType(_CassandraType):
    """CQL float type (32-bit IEEE 754 floating point)."""

class DoubleType(_CassandraType):
    """CQL double type (64-bit IEEE 754 floating point)."""

class UTF8Type(_CassandraType):
    """CQL text type (UTF-8 encoded strings)."""

VarcharType = UTF8Type
"""Alias for UTF8Type (CQL varchar type)."""

class InetAddressType(_CassandraType):
    """CQL inet type for IP addresses."""

class AsciiType(_CassandraType):
    """CQL ascii type for ASCII-only strings."""

Time-Related Types

Types for handling temporal data with proper timezone and precision support.

class DateType(_CassandraType):
    """Base class for date-related types."""

class TimestampType(DateType):
    """CQL timestamp type (datetime with millisecond precision)."""

class SimpleDateType(DateType):
    """CQL date type (date without time component)."""

class TimeType(_CassandraType):
    """CQL time type (time of day with nanosecond precision)."""

class TimeUUIDType(UUIDType):
    """CQL timeuuid type (time-based UUID version 1)."""

Collection Types

Types for handling CQL collections with proper nesting and type safety.

class ListType(_ParameterizedType):
    """CQL list type (ordered collection)."""
    
    def __init__(self, subtype, frozen=False):
        """
        Parameters:
        - subtype (_CassandraType): Type of list elements
        - frozen (bool): Whether the collection is frozen
        """

class SetType(_ParameterizedType):
    """CQL set type (unordered collection of unique elements)."""
    
    def __init__(self, subtype, frozen=False):
        """
        Parameters:
        - subtype (_CassandraType): Type of set elements  
        - frozen (bool): Whether the collection is frozen
        """

class MapType(_ParameterizedType):
    """CQL map type (key-value collection)."""
    
    def __init__(self, key_type, value_type, frozen=False):
        """
        Parameters:
        - key_type (_CassandraType): Type of map keys
        - value_type (_CassandraType): Type of map values
        - frozen (bool): Whether the collection is frozen
        """

class TupleType(_ParameterizedType):
    """CQL tuple type (fixed-size collection of heterogeneous elements)."""
    
    def __init__(self, subtypes):
        """
        Parameters:
        - subtypes (list): List of types for tuple elements
        """

Complex Types

Advanced types for user-defined structures and legacy support.

class UserType(_ParameterizedType):
    """CQL user-defined type (UDT)."""
    
    def __init__(self, keyspace, typename, field_names, field_types):
        """
        Parameters:
        - keyspace (str): Keyspace containing the UDT
        - typename (str): Name of the user-defined type
        - field_names (list): Names of UDT fields
        - field_types (list): Types of UDT fields
        """
    
    @property
    def keyspace(self):
        """str: Keyspace containing this UDT"""
    
    @property
    def typename(self):
        """str: Name of this UDT"""
    
    @property
    def field_names(self):
        """list: Names of fields in this UDT"""
    
    @property
    def field_types(self):
        """list: Types of fields in this UDT"""

class CompositeType(_CassandraType):
    """Legacy composite type (deprecated)."""
    
    def __init__(self, subtypes):
        """
        Parameters:
        - subtypes (list): Component types
        """

class DynamicCompositeType(_CassandraType):
    """Legacy dynamic composite type (deprecated)."""

class CounterColumnType(_CassandraType):
    """CQL counter type for counter columns."""

class ReversedType(_CassandraType):
    """Wrapper type for reversed sort order."""
    
    def __init__(self, subtype):
        """
        Parameters:
        - subtype (_CassandraType): The wrapped type
        """

class FrozenType(_CassandraType):
    """Wrapper type for frozen collections and UDTs."""
    
    def __init__(self, subtype):
        """
        Parameters:
        - subtype (_CassandraType): The frozen type
        """

Type Utilities

Utility functions for type lookup, parsing, and validation.

def lookup_casstype(casstype_name):
    """
    Look up a CQL type by name.
    
    Parameters:
    - casstype_name (str): Name of the CQL type
    
    Returns:
    _CassandraType: Type class for the given name
    
    Raises:
    - UnknownTypeException: If type name is not recognized
    """

def parse_casstype_args(typestring):
    """
    Parse type arguments from a CQL type string.
    
    Parameters:
    - typestring (str): CQL type string (e.g., "map<text,int>")
    
    Returns:
    list: Parsed type arguments
    """

def is_counter_type(casstype):
    """
    Check if a type is a counter type.
    
    Parameters:
    - casstype (_CassandraType): Type to check
    
    Returns:
    bool: True if the type is a counter type
    """

def cql_typename(casstype_name):
    """
    Get the CQL name for a type.
    
    Parameters:
    - casstype_name (str): Internal type name
    
    Returns:
    str: CQL type name
    """

Time and UUID Utilities

Utility functions for working with time-based data and UUIDs.

def unix_time_from_uuid1(uuid_arg):
    """
    Extract Unix timestamp from a time-based UUID (version 1).
    
    Parameters:
    - uuid_arg (UUID): Time-based UUID
    
    Returns:
    float: Unix timestamp in seconds
    """

def datetime_from_timestamp(timestamp):
    """
    Convert a timestamp to a datetime object.
    
    Parameters:
    - timestamp (int or float): Unix timestamp
    
    Returns:
    datetime: Corresponding datetime object
    """

def datetime_from_uuid1(uuid_arg):
    """
    Extract datetime from a time-based UUID (version 1).
    
    Parameters:
    - uuid_arg (UUID): Time-based UUID
    
    Returns:
    datetime: Datetime when the UUID was created
    """

def min_uuid_from_time(timestamp):
    """
    Generate the minimum possible UUID for a given timestamp.
    
    Parameters:
    - timestamp (datetime or float): Time for UUID generation
    
    Returns:
    UUID: Minimum UUID for the given time
    """

def max_uuid_from_time(timestamp):
    """
    Generate the maximum possible UUID for a given timestamp.
    
    Parameters:
    - timestamp (datetime or float): Time for UUID generation
    
    Returns:
    UUID: Maximum UUID for the given time
    """

def uuid_from_time(time_arg, node=None, clock_seq=None):
    """
    Generate a time-based UUID (version 1).
    
    Parameters:
    - time_arg (datetime or float): Time for UUID generation
    - node (int): MAC address (6 bytes) or None for random
    - clock_seq (int): Clock sequence or None for random
    
    Returns:
    UUID: Generated time-based UUID
    """

Usage Examples

Working with Basic Types

from cassandra.cqltypes import *
from decimal import Decimal
import uuid
from datetime import datetime, date, time

# Working with various primitive types
session.execute("""
    CREATE TABLE type_examples (
        id uuid PRIMARY KEY,
        text_val text,
        int_val int,
        bigint_val bigint,
        float_val float,
        double_val double,
        decimal_val decimal,
        bool_val boolean,
        blob_val blob,
        inet_val inet,
        timestamp_val timestamp,
        date_val date,
        time_val time,
        timeuuid_val timeuuid
    )
""")

# Insert data with various types
session.execute("""
    INSERT INTO type_examples (
        id, text_val, int_val, bigint_val, float_val, double_val,
        decimal_val, bool_val, blob_val, inet_val, timestamp_val,
        date_val, time_val, timeuuid_val
    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [
    uuid.uuid4(),
    'Hello, Cassandra!',
    42,
    9223372036854775807,
    3.14159,
    2.718281828459045,
    Decimal('123.456'),
    True,
    b'binary data',
    '192.168.1.1',
    datetime.now(),
    date.today(),
    time(14, 30, 0),
    uuid.uuid1()  # Time-based UUID
])

Working with Collections

# Create table with collection types
session.execute("""
    CREATE TABLE collection_examples (
        id uuid PRIMARY KEY,
        tags set<text>,
        scores list<int>,
        metadata map<text, text>,
        coordinates tuple<double, double>
    )
""")

# Insert collection data
session.execute("""
    INSERT INTO collection_examples (id, tags, scores, metadata, coordinates)
    VALUES (?, ?, ?, ?, ?)
""", [
    uuid.uuid4(),
    {'python', 'cassandra', 'database'},  # Set
    [95, 87, 92, 88],                     # List
    {'author': 'Alice', 'version': '1.0'}, # Map
    (40.7128, -74.0060)                   # Tuple (NYC coordinates)
])

# Working with nested collections (requires frozen)
session.execute("""
    CREATE TABLE nested_collections (
        id uuid PRIMARY KEY,
        matrix list<frozen<list<int>>>,
        user_scores map<text, frozen<list<int>>>
    )
""")

session.execute("""
    INSERT INTO nested_collections (id, matrix, user_scores) 
    VALUES (?, ?, ?)
""", [
    uuid.uuid4(),
    [[1, 2, 3], [4, 5, 6], [7, 8, 9]],  # List of lists
    {
        'alice': [95, 87, 92],
        'bob': [88, 91, 85]
    }
])

Working with User-Defined Types

# Create a user-defined type
session.execute("""
    CREATE TYPE address (
        street text,
        city text,
        state text,
        zip_code text
    )
""")

session.execute("""
    CREATE TABLE users_with_address (
        id uuid PRIMARY KEY,
        name text,
        home_address address,
        work_address address
    )
""")

# Insert data with UDT
from collections import namedtuple

# Create a named tuple that matches the UDT structure
Address = namedtuple('Address', ['street', 'city', 'state', 'zip_code'])

home = Address('123 Main St', 'Anytown', 'CA', '12345')
work = Address('456 Business Ave', 'Corporate City', 'NY', '67890')

session.execute("""
    INSERT INTO users_with_address (id, name, home_address, work_address)
    VALUES (?, ?, ?, ?)
""", [uuid.uuid4(), 'Alice Smith', home, work])

# Query and access UDT fields
result = session.execute("SELECT * FROM users_with_address")
for row in result:
    print(f"User: {row.name}")
    print(f"Home: {row.home_address.street}, {row.home_address.city}")
    print(f"Work: {row.work_address.street}, {row.work_address.city}")

Time and UUID Operations

from cassandra.util import *
import uuid
from datetime import datetime, timedelta

# Generate time-based UUIDs
now = datetime.now()
uuid1_now = uuid_from_time(now)
uuid1_past = uuid_from_time(now - timedelta(hours=1))

print(f"UUID for now: {uuid1_now}")
print(f"UUID for 1 hour ago: {uuid1_past}")

# Extract time from UUID
extracted_time = datetime_from_uuid1(uuid1_now)
timestamp = unix_time_from_uuid1(uuid1_now)

print(f"Extracted datetime: {extracted_time}")
print(f"Unix timestamp: {timestamp}")

# Generate min/max UUIDs for time range queries
start_time = datetime.now() - timedelta(days=1)
end_time = datetime.now()

min_uuid = min_uuid_from_time(start_time)
max_uuid = max_uuid_from_time(end_time)

# Query using UUID time range
session.execute("""
    SELECT * FROM time_series 
    WHERE id >= ? AND id <= ?
""", [min_uuid, max_uuid])

# Convert timestamps
timestamp = 1640995200  # Example timestamp
dt = datetime_from_timestamp(timestamp)
print(f"Timestamp {timestamp} -> {dt}")

Type Introspection and Custom Handling

from cassandra.cqltypes import lookup_casstype, parse_casstype_args

# Look up types by name
text_type = lookup_casstype('text')
int_type = lookup_casstype('int')
uuid_type = lookup_casstype('uuid')

print(f"Text type: {text_type}")
print(f"Int type: {int_type}")
print(f"UUID type: {uuid_type}")

# Parse complex type strings
map_args = parse_casstype_args('map<text,int>')
list_args = parse_casstype_args('list<uuid>')
tuple_args = parse_casstype_args('tuple<text,int,boolean>')

print(f"Map args: {map_args}")
print(f"List args: {list_args}")  
print(f"Tuple args: {tuple_args}")

# Custom type handling in results
def custom_decimal_handler(value):
    """Convert Decimal to float for JSON serialization"""
    if isinstance(value, Decimal):
        return float(value)
    return value

# Apply custom handling to query results
result = session.execute("SELECT decimal_column FROM my_table")
for row in result:
    processed_value = custom_decimal_handler(row.decimal_column)
    print(f"Processed decimal: {processed_value}")

Working with Null and Empty Values

from cassandra.cqltypes import EmptyValue

# Insert null/empty values
session.execute("""
    INSERT INTO nullable_table (id, optional_text, optional_int)
    VALUES (?, ?, ?)
""", [uuid.uuid4(), None, None])  # None becomes CQL null

# Insert using EmptyValue explicitly
session.execute("""
    INSERT INTO nullable_table (id, optional_text, optional_int)
    VALUES (?, ?, ?)
""", [uuid.uuid4(), EmptyValue(), EmptyValue()])

# Check for null values in results
result = session.execute("SELECT * FROM nullable_table")
for row in result:
    if row.optional_text is None:
        print("Text field is null")
    if row.optional_int is None:
        print("Int field is null")

Type Validation and Conversion

from cassandra.cqltypes import *
import uuid
from decimal import Decimal

def validate_and_convert_data(data, expected_types):
    """Validate and convert data according to expected CQL types"""
    converted = {}
    
    for field, value in data.items():
        expected_type = expected_types[field]
        
        if isinstance(expected_type, UTF8Type):
            converted[field] = str(value) if value is not None else None
        elif isinstance(expected_type, Int32Type):
            converted[field] = int(value) if value is not None else None
        elif isinstance(expected_type, UUIDType):
            if isinstance(value, str):
                converted[field] = uuid.UUID(value)
            else:
                converted[field] = value
        elif isinstance(expected_type, DecimalType):
            if isinstance(value, (int, float, str)):
                converted[field] = Decimal(str(value))
            else:
                converted[field] = value
        else:
            converted[field] = value
    
    return converted

# Example usage
schema = {
    'id': UUIDType(),
    'name': UTF8Type(),
    'age': Int32Type(),
    'balance': DecimalType()
}

raw_data = {
    'id': 'f47ac10b-58cc-4372-a567-0e02b2c3d479',
    'name': 'Alice',
    'age': '30',  # String that should be int
    'balance': '123.45'  # String that should be Decimal
}

validated_data = validate_and_convert_data(raw_data, schema)
print(f"Validated data: {validated_data}")

# Use validated data in query
session.execute("""
    INSERT INTO users (id, name, age, balance) VALUES (?, ?, ?, ?)
""", [validated_data['id'], validated_data['name'], 
      validated_data['age'], validated_data['balance']])

Install with Tessl CLI