Python legacy driver versions

Connect to Astra from Python using a previous version of the DataStax Driver. Before proceeding please consider upgrading to the most current driver if possible. However, if you cannot upgrade, or are using a Framework that doesn’t support Astra secure connect, the following instructions will help you get connected.

Prerequisites

Procedure

  1. Copy the certificate files from the Astra secure connect bundle to ../secure-connect/. This folder should contain the ca.cert, cert, and key files.

  1. Replace the <hostname> with your Astra database hostname.

  1. Replace the <port> with your Astra database port.

  1. Replace <username> and <password> with the Astra database Client ID and Client Secret.

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from ssl import SSLContext, PROTOCOL_TLSv1, CERT_REQUIRED

ssl_context = SSLContext(PROTOCOL_TLSv1)
ssl_context.load_verify_locations('../secure-connect/ca.crt')
ssl_context.verify_mode = CERT_REQUIRED
ssl_context.load_cert_chain(
    certfile='../secure-connect/cert',
    keyfile='../secure-connect/key')


auth_provider = PlainTextAuthProvider('<username>', '<password>')
cluster = Cluster(['<hostname>'], port=<port>, ssl_context=ssl_context, auth_provider=auth_provider)
session = cluster.connect()

row = session.execute("select release_version from system.local").one()
if row:
  print(row[0])
else:
  print("An error occurred.")


print(cluster.metadata.keyspaces)

Example

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from ssl import SSLContext, PROTOCOL_TLSv1, CERT_REQUIRED

ssl_context = SSLContext(PROTOCOL_TLSv1)
ssl_context.load_verify_locations('../secure-connect/ca.crt')
ssl_context.verify_mode = CERT_REQUIRED
ssl_context.load_cert_chain(
    certfile='../secure-connect/cert',
    keyfile='../secure-connect/key')


auth_provider = PlainTextAuthProvider('john.smith', 'jsP@ssw0rd')
cluster = Cluster(['7bb9cd7a-e49d-49a6-aa3d-be4878f974ef-us-east1.db.astra.datastax.com'], port=31575, ssl_context=ssl_context, auth_provider=auth_provider)
session = cluster.connect()

row = session.execute("select release_version from system.local").one()
if row:
  print(row[0])
else:
  print("An error occurred.")


print(cluster.metadata.keyspaces)

In the example above, the following variables are used:

  • hostname is 7bb9cd7a-e49d-49a6-aa3d-be4878f974ef-us-east1.db.astra.datastax.com

  • port is 31575

  • username is john.smith

  • password is jsP@ssw0rd

Flask CQLAlchemy

Following the procedure above this code allows CQLAlchemy to utilize Astra.

Flask CQLAlchemy

import uuid
from flask import Flask
from flask_cqlalchemy import CQLAlchemy
from ssl import SSLContext, PROTOCOL_TLS, CERT_REQUIRED
from cassandra.auth import PlainTextAuthProvider


auth_provider = PlainTextAuthProvider(username='<username>', password='<password>')
ssl_context = SSLContext(PROTOCOL_TLSv1)
ssl_context.load_verify_locations("../secure-connect/ca.crt")
ssl_context.verify_mode = CERT_REQUIRED
ssl_context.load_cert_chain(
  certfile="../secure-connect/cert",
  keyfile="secure-connect/key")

app = Flask(__name__)
app.config['CASSANDRA_HOSTS'] = ['<hostname>']
app.config['CASSANDRA_SETUP_KWARGS'] = dict(ssl_context=ssl_context, port="<port>", auth_provider=auth_provider)
app.config['CASSANDRA_KEYSPACE'] = "cqlengine"
db = CQLAlchemy(app)


class User(db.Model):
    uid = db.columns.UUID(primary_key=True, default=uuid.uuid4)
    username = db.columns.Text(required=False)

Example

import uuid
from flask import Flask
from flask_cqlalchemy import CQLAlchemy
from ssl import SSLContext, PROTOCOL_TLS, CERT_REQUIRED
from cassandra.auth import PlainTextAuthProvider


auth_provider = PlainTextAuthProvider(username='jane.smith', password='jsP@ssw0rd')
ssl_context = SSLContext(PROTOCOL_TLSv1)
ssl_context.load_verify_locations("../secure-connect/ca.crt")
ssl_context.verify_mode = CERT_REQUIRED
ssl_context.load_cert_chain(
  certfile="../secure-connect/cert",
  keyfile="secure-connect/key")

app = Flask(__name__)
app.config['CASSANDRA_HOSTS'] = ['7bb9cd7a-e49d-49a6-aa3d-be4878f974ef-us-east1.db.astra.datastax.com']
app.config['CASSANDRA_SETUP_KWARGS'] = dict(ssl_context=ssl_context, port="31575", auth_provider=auth_provider)
app.config['CASSANDRA_KEYSPACE'] = "cqlengine"
db = CQLAlchemy(app)


class User(db.Model):
    uid = db.columns.UUID(primary_key=True, default=uuid.uuid4)
    username = db.columns.Text(required=False)