Python legacy driver
Connect to Astra DB from Python using a previous version of the DataStax Driver.
If possible, upgrade to a Python Native driver, which is version 3.24 or greater. For more, see Drivers for Astra. However, if you cannot upgrade, or are using a Framework that doesn’t support Astra DB secure connect, the following instructions will help you get connected. |
Prerequisites
-
Ensure the Python driver is installed. For more, see DataStax Python Driver Installation.
-
Collect the required information for Connecting with legacy drivers.
Procedure
-
Copy the certificate files from the Astra DB secure connect bundle to
../secure-connect/
.This folder should contain the
ca.cert
,cert
, andkey
files. -
Replace the
<hostname>
with your Astra DB database hostname. -
Replace the
<port>
with your Astra DB database port. -
Replace
<username>
and<password>
with the Astra DB 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
is7bb9cd7a-e49d-49a6-aa3d-be4878f974ef-us-east1.db.astra.datastax.com
-
port
is31575
-
username
isjohn.smith
-
password
isjsP@ssw0rd
Flask CQLAlchemy framework
Following the procedure above this code allows CQLAlchemy to utilize Astra DB.
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)