Cloud Functions (Python Driver)

Connecting to Astra DB from a Google Cloud Function using Python Driver and Cloud Console

🏠 Back to home | Written by Artem Chebotko

📋 On this page

• A - Overview
• B - Prerequisites
• C - Creating a Google Cloud Function to Access Astra DB using Python Driver and Cloud Console

A - Overview

Cloud Functions is Google's function-as-a-service offering that provides a serverless execution environment for your code. Cloud Functions are commonly used to:

• Extend Astra DB with additional data processing capabilities, such as aggregating, summarizing and validating data periodically;
• Connect Astra DB with other cloud services into data pipelines that move, process and analyze data.

B - Prerequisites

• Create an Astra Database
• Create an Astra Token
• Download a Secure Connect Bundle
• Optionally, if you are new to Cloud Functions, practice creating a simpler function first

C - Creating a Google Cloud Function to Access Astra DB using Python Driver and Cloud Console

✅ Step 1: Create a secret with the secure connect bundle file.

1. Go to the Secret Manager page, select a project that has Secret Manager and Cloud Functions enabled, and click Create secret.
2. Give a Name to the secret and upload the secure connect bundle file as a Secret value. (See the Prerequisites section above if you need to download your secure connect bundle.) Optionally, customize other secret management settings.

3. Click Create secret.

4. On the Secret Manager page, find the newly created secret.

✅ Step 2: Create a function.

1. Go to the Functions Overview page, select the same project that has Secret Manager and Cloud Functions enabled, and click Create function.
2. Under the Basics section, specify preferred Function name and Region.
3. Under the Trigger section, select HTTP, Allow unauthenticated invocations, and Require HTTPS.

4. Click Save.

5. Under the Runtime, build, connections and security settings section, customize additional settings and create these Runtime environment variables:

• ASTRA_DB_CLIENT_ID: A Client ID is generated together with an application token (see the Prerequisites section above).
• ASTRA_DB_CLIENT_SECRET: A Client secret is generated together with an application token (see the Prerequisites section above).

Note that, for better security, you can alternatively use the Secret Manager service to store and manage a client secret. A secret can then be similarly exposed as an environment variable. The settings can be found under the Runtime, build, connections and security settings section, the Security tab, and the Secrets field.

6. Under the Runtime, build, connections and security settings section and the Security, click Reference a secret. Select the previously created Secret with the secure connect bundle file, Grant the service account access to the secret, if needed, use Mounted as volume in the Reference method field, and enter secrets in the Mount path field.

Notice the final Path that should be used to access the secure connect bundle in the function code.

7. Click Done and Next.

8. Select Python 3.7 or your preferred version in the Runtime field.

9. Select Inline Editor in the Source code field.

10. Enter query_astra_db in the Entry point field.

11. Add cassandra-driver, a Python client library for Apache Cassandra, DataStax Astra DB and DataStax Enterprise, to the requirements.txt file.

12. Replace the main.py content with:

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
import os
from shutil import copyfile

def query_astra_db(request):

    # Copy the secure connect bundle file to the writable part of the file system /tmp
    copyfile('/secrets/secure-connect-secret', '/tmp/secure-connect-for-my-database.zip')

    ASTRA_DB_CLIENT_ID = os.environ.get('ASTRA_DB_CLIENT_ID')
    ASTRA_DB_CLIENT_SECRET = os.environ.get('ASTRA_DB_CLIENT_SECRET')

    cloud_config= {
          'secure_connect_bundle': '/tmp/secure-connect-for-my-database.zip'
    }
    auth_provider = PlainTextAuthProvider(ASTRA_DB_CLIENT_ID, ASTRA_DB_CLIENT_SECRET)
    cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider, protocol_version=4)
    session = cluster.connect()

    row = session.execute("SELECT cql_version FROM system.local WHERE key = 'local';").one()
    print(row[0])

    print ('Success')

You can learn more about the code above by reading the cassandra-driver documentation.

✅ Step 3: Deploy the function.

1. Click Deploy.

2. On the Cloud Functions Overview page, find the newly deployed function.

✅ Step 4: Test the function.

1. Under Actions, select Test function.

2. On the testing page, click Test the function and observe the output.

Notice the CQL version output 3.4.5 and status code 200.

✅ Step 5: View logs.

You can further explore the log history by either clicking on the Logs tab or the View all logs link that opens Logs Explorer.