Cassandra Python Driver

The DataStax Python Driver is the official library for connecting Python applications to Apache Cassandra. It provides a Cluster object that manages connections, a Session object for executing queries, and support for prepared statements, asynchronous execution, and result set mapping to Python objects.

Installation

pip install cassandra-driver

For better performance, install the optional C extensions:

pip install cassandra-driver[speedups]

Connecting to Cassandra

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

auth = PlainTextAuthProvider(username='alice', password='Al1ce$ecure!')

cluster = Cluster(
  contact_points=['127.0.0.1'],
  port=9042,
  auth_provider=auth
)

session = cluster.connect('ecommerce')
print("Connected to keyspace:", session.keyspace)

Multi-Node Connection

cluster = Cluster(
  contact_points=['10.0.0.1', '10.0.0.2', '10.0.0.3'],
  port=9042,
  auth_provider=auth
)
session = cluster.connect('ecommerce')

Executing Simple Queries

rows = session.execute("SELECT product_id, name, price FROM products LIMIT 10")

for row in rows:
  print(f"Product: {row.product_id} | {row.name} | ${row.price:.2f}")

Prepared Statements (Best Practice)

Prepare all frequently-used CQL statements at startup. The driver sends the statement text once, and all subsequent executions only send bound parameter values — faster and safer.

# Prepare at startup:
insert_product = session.prepare("""
  INSERT INTO products (product_id, name, price, category)
  VALUES (?, ?, ?, ?)
""")

select_by_cat = session.prepare("""
  SELECT product_id, name, price
  FROM products_by_category
  WHERE category = ?
  ORDER BY price ASC
  LIMIT 20
""")

# Execute with bound values:
from uuid import uuid4
from decimal import Decimal

session.execute(insert_product, (uuid4(), 'Bluetooth Speaker', Decimal('49.99'), 'Electronics'))

rows = session.execute(select_by_cat, ('Electronics',))
for row in rows:
  print(row.name, row.price)

Consistency Level per Query

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement

stmt = SimpleStatement(
  "SELECT * FROM orders_by_customer WHERE customer_id = %s",
  consistency_level=ConsistencyLevel.LOCAL_QUORUM
)

rows = session.execute(stmt, (customer_id,))

Asynchronous Queries

# Fire and forget (async):
future = session.execute_async(insert_product,
  (uuid4(), 'HDMI Cable', Decimal('14.99'), 'Electronics'))

# Do other work while query executes...

# Wait for result when needed:
try:
  result = future.result()
  print("Insert successful")
except Exception as e:
  print(f"Insert failed: {e}")

Async Batch Ingestion

from concurrent.futures import as_completed

products = [
  (uuid4(), 'USB Hub',      Decimal('29.99'), 'Electronics'),
  (uuid4(), 'Mouse Pad',    Decimal(' 9.99'), 'Accessories'),
  (uuid4(), 'Laptop Stand', Decimal('39.99'), 'Accessories'),
]

futures = [session.execute_async(insert_product, p) for p in products]

for future in futures:
  try:
    future.result()
  except Exception as e:
    print(f"Error: {e}")

Handling Large Result Sets with Paging

from cassandra.query import SimpleStatement

stmt = SimpleStatement(
  "SELECT * FROM products_by_category WHERE category = 'Electronics'",
  fetch_size=50   # 50 rows per page
)

# Driver fetches pages automatically as you iterate:
for row in session.execute(stmt):
  print(row.name, row.price)

Named Tuple and Dict Results

# Default: returns named tuples (access by column name)
row = session.execute("SELECT name, price FROM products WHERE product_id = %s",
                      (product_id,)).one()
print(row.name)   # named tuple access
print(row.price)

# Convert to dictionary:
row_dict = row._asdict()
print(row_dict['name'])

Object Mapping with cqlengine

The driver includes cqlengine, an ORM-style mapper that lets you define Cassandra tables as Python classes.

from cassandra.cqlengine import columns
from cassandra.cqlengine.models import Model
from cassandra.cqlengine.management import sync_table
from cassandra.cqlengine import connection

connection.setup(['127.0.0.1'], 'ecommerce', protocol_version=4)

class Product(Model):
  __keyspace__ = 'ecommerce'
  product_id = columns.UUID(primary_key=True, default=uuid4)
  name       = columns.Text()
  price      = columns.Decimal()
  category   = columns.Text()

# Create table if not exists:
sync_table(Product)

# Insert:
Product.create(name='Mechanical Keyboard', price=Decimal('89.99'), category='Electronics')

# Query:
products = Product.objects.filter(product_id=some_uuid)
for p in products:
  print(p.name, p.price)

Batch Statements

from cassandra.query import BatchStatement, BatchType

batch = BatchStatement(batch_type=BatchType.LOGGED)
batch.add(insert_product, (uuid4(), 'Webcam', Decimal('59.99'), 'Electronics'))
batch.add(insert_product, (uuid4(), 'Ring Light', Decimal('34.99'), 'Accessories'))
session.execute(batch)

Closing Connections Properly

session.shutdown()
cluster.shutdown()

Use a context manager or try/finally block in long-running scripts to ensure clean shutdown.

Error Handling

from cassandra import WriteTimeout, ReadTimeout, Unavailable

try:
  session.execute(insert_product, (uuid4(), 'Tablet', Decimal('299.99'), 'Electronics'))
except WriteTimeout as e:
  print(f"Write timed out: {e}")
  # Write may have partially succeeded — check before retrying
except Unavailable as e:
  print(f"Not enough replicas: {e}")
except Exception as e:
  print(f"Unexpected error: {e}")

Connection Configuration Reference

Parameter              Purpose                          Default
──────────────────────────────────────────────────────────────────
contact_points         Initial nodes to contact         ['127.0.0.1']
port                   CQL native transport port        9042
auth_provider          Authentication credentials       None
load_balancing_policy  How to pick replica nodes        DC-aware round robin
default_consistency    Default consistency level        LOCAL_ONE
connect_timeout        Connection establishment timeout 5 seconds
request_timeout        Individual query timeout         10 seconds

Summary

The DataStax Python Driver connects Python applications to Cassandra with minimal setup. Use the Cluster class to manage connections and the Session class to execute queries. Always use prepared statements for repeated queries to save parsing overhead. Use execute_async for concurrent writes, automatic paging for large result sets, and cqlengine for ORM-style data access. Handle WriteTimeout carefully — it does not always mean the write failed, since some replicas may have accepted it.