Getting Started¶

Connecting to Cassandra¶

Before we can start executing any queries against a Cassandra cluster we need to setup an instance of Cluster. As the name suggests, you will typically have one instance of Cluster for each Cassandra cluster you want to interact with.

The simplest way to create a Cluster is like this: First, make sure you have the Cassandra driver properly installed.

from cassandra.cluster import Cluster

cluster = Cluster()

This will attempt to connection to a Cassandra instance on your local machine (127.0.0.1). You can also specify a list of IP addresses for nodes in your cluster:

from cassandra.cluster import Cluster

cluster = Cluster(['192.168.0.1', '192.168.0.2'])

The set of IP addresses we pass to the Cluster is simply an initial set of contact points. After the driver connects to one of these nodes it will automatically discover the rest of the nodes in the cluster and connect to them, so you don’t need to list every node in your cluster.

If you need to use a non-standard port, use SSL, or customize the driver’s behavior in some other way, this is the place to do it:

from cassandra.cluster import Cluster
cluster = Cluster(['192.168.0.1', '192.168.0.2'], port=..., ssl_context=...)

Instantiating a Cluster does not actually connect us to any nodes. To establish connections and begin executing queries we need a Session, which is created by calling Cluster.connect():

cluster = Cluster()
session = cluster.connect()

The connect() method takes an optional keyspace argument which sets the default keyspace for all queries made through that Session:

cluster = Cluster()
session = cluster.connect('mykeyspace')

You can always change a Session’s keyspace using set_keyspace() or by executing a USE <keyspace> query:

session.set_keyspace('users')
# or you can do this instead
session.execute('USE users')

Profiles are passed in by execution_profiles dict.

In this case we can construct the base ExecutionProfile passing all attributes:

from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import WhiteListRoundRobinPolicy, DowngradingConsistencyRetryPolicy
from cassandra.query import tuple_factory

profile = ExecutionProfile(
    load_balancing_policy=WhiteListRoundRobinPolicy(['127.0.0.1']),
    retry_policy=DowngradingConsistencyRetryPolicy(),
    consistency_level=ConsistencyLevel.LOCAL_QUORUM,
    serial_consistency_level=ConsistencyLevel.LOCAL_SERIAL,
    request_timeout=15,
    row_factory=tuple_factory
)
cluster = Cluster(execution_profiles={EXEC_PROFILE_DEFAULT: profile})
session = cluster.connect()

print(session.execute("SELECT release_version FROM system.local").one())

Users are free to setup additional profiles to be used by name:

profile_long = ExecutionProfile(request_timeout=30)
cluster = Cluster(execution_profiles={'long': profile_long})
session = cluster.connect()
session.execute(statement, execution_profile='long')

Also, parameters passed to Session.execute or attached to Statements are still honored as before.

Executing Queries¶

Now that we have a Session we can begin to execute queries. The simplest way to execute a query is to use execute():

rows = session.execute('SELECT name, age, email FROM users')
for user_row in rows:
    print user_row.name, user_row.age, user_row.email

This will transparently pick a Cassandra node to execute the query against and handle any retries that are necessary if the operation fails.

By default, each row in the result set will be a namedtuple. Each row will have a matching attribute for each column defined in the schema, such as name, age, and so on. You can also treat them as normal tuples by unpacking them or accessing fields by position. The following three examples are equivalent:

rows = session.execute('SELECT name, age, email FROM users')
for row in rows:
    print row.name, row.age, row.email

rows = session.execute('SELECT name, age, email FROM users')
for (name, age, email) in rows:
    print name, age, email

rows = session.execute('SELECT name, age, email FROM users')
for row in rows:
    print row[0], row[1], row[2]

If you prefer another result format, such as a dict per row, you can change the row_factory attribute.

As mentioned in our Drivers Best Practices Guide, it is highly recommended to use Prepared statements for your frequently run queries.

Prepared Statements¶

Prepared statements are queries that are parsed by Cassandra and then saved for later use. When the driver uses a prepared statement, it only needs to send the values of parameters to bind. This lowers network traffic and CPU utilization within Cassandra because Cassandra does not have to re-parse the query each time.

To prepare a query, use Session.prepare():

user_lookup_stmt = session.prepare("SELECT * FROM users WHERE user_id=?")

users = []
for user_id in user_ids_to_query:
    user = session.execute(user_lookup_stmt, [user_id])
    users.append(user)

prepare() returns a PreparedStatement instance which can be used in place of SimpleStatement instances or literal string queries. It is automatically prepared against all nodes, and the driver handles re-preparing against new nodes and restarted nodes when necessary.

Note that the placeholders for prepared statements are ? characters. This is different than for simple, non-prepared statements (although future versions of the driver may use the same placeholders for both).

session.execute(
    """
    INSERT INTO users (name, credits, user_id)
    VALUES (%s, %s, %s)
    """,
    ("John O'Reilly", 42, uuid.uuid1())
)

INSERT INTO users (name, credits, user_id)
VALUES ('John O''Reilly', 42, 2644bada-852c-11e3-89fb-e0b9a54a6d93)

session.execute("INSERT INTO USERS (name, age) VALUES (%s, %d)", ("bob", 42))  # wrong

session.execute("INSERT INTO foo (bar) VALUES (%s)", "blah")  # wrong
session.execute("INSERT INTO foo (bar) VALUES (%s)", ("blah"))  # wrong
session.execute("INSERT INTO foo (bar) VALUES (%s)", ("blah", ))  # right
session.execute("INSERT INTO foo (bar) VALUES (%s)", ["blah"])  # right

session.execute(
    """
    INSERT INTO users (name, credits, user_id, username)
    VALUES (%(name)s, %(credits)s, %(user_id)s, %(name)s)
    """,
    {'name': "John O'Reilly", 'credits': 42, 'user_id': uuid.uuid1()}
)

from cassandra import ReadTimeout

query = "SELECT * FROM users WHERE user_id=%s"
future = session.execute_async(query, [user_id])

# ... do some other work

try:
    rows = future.result()
    user = rows[0]
    print user.name, user.age
except ReadTimeout:
    log.exception("Query timed out:")

# build a list of futures
futures = []
query = "SELECT * FROM users WHERE user_id=%s"
for user_id in ids_to_fetch:
    futures.append(session.execute_async(query, [user_id])

# wait for them to complete and use the results
for future in futures:
    rows = future.result()
    print rows[0].name

def handle_success(rows):
    user = rows[0]
    try:
        process_user(user.name, user.age, user.id)
    except Exception:
        log.error("Failed to process user %s", user.id)
        # don't re-raise errors in the callback

def handle_error(exception):
    log.error("Failed to fetch user info: %s", exception)


future = session.execute_async(query)
future.add_callbacks(handle_success, handle_error)

Setting a Consistency Level¶

The consistency level used for a query determines how many of the replicas of the data you are interacting with need to respond for the query to be considered a success.

By default, ConsistencyLevel.LOCAL_ONE will be used for all queries. You can specify a different default by setting the ExecutionProfile.consistency_level for the execution profile with key EXEC_PROFILE_DEFAULT. To specify a different consistency level per request, wrap queries in a SimpleStatement:

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement

query = SimpleStatement(
    "INSERT INTO users (name, age) VALUES (%s, %s)",
    consistency_level=ConsistencyLevel.QUORUM)
session.execute(query, ('John', 42))

from cassandra import ConsistencyLevel

cluster = Cluster()
session = cluster.connect("mykeyspace")
user_lookup_stmt = session.prepare("SELECT * FROM users WHERE user_id=?")
user_lookup_stmt.consistency_level = ConsistencyLevel.QUORUM

# these will both use QUORUM
user1 = session.execute(user_lookup_stmt, [user_id1])[0]
user2 = session.execute(user_lookup_stmt, [user_id2])[0]

# override the QUORUM default
user3_lookup = user_lookup_stmt.bind([user_id3])
user3_lookup.consistency_level = ConsistencyLevel.ALL
user3 = session.execute(user3_lookup)

from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import ConstantSpeculativeExecutionPolicy
from cassandra.query import SimpleStatement

# Configure the speculative execution policy
ep = ExecutionProfile(
    speculative_execution_policy=ConstantSpeculativeExecutionPolicy(delay=.5, max_attempts=10)
)
cluster = Cluster(..., execution_profiles={EXEC_PROFILE_DEFAULT: ep})
session = cluster.connect()

# Mark the query idempotent
query = SimpleStatement(
    "UPDATE my_table SET list_col = [1] WHERE pk = 1",
    is_idempotent=True
)

# Execute. A new query will be sent to the server every 0.5 second
# until we receive a response, for a max number attempts of 10.
session.execute(query)