Astrapy Python Client
Astrapy is a Python client library for interacting with the Data API on DataStax Astra DB. It provides a straightforward interface for working with both document collections (JSON) and tables. Currently at version 2.2.1, the library sees active development with frequent minor releases introducing new features and improvements.
Warnings
- breaking Version 2.0.0 introduced significant API changes, including full support for Tables, a major revision of the overall interface, and new astrapy-specific data types (e.g., `DataAPIVector`, `DataAPIDate`). Code written for 1.x versions will likely require substantial updates.
- breaking Python 3.8 support was dropped in version 2.2.0. The library now requires Python 3.9 or higher.
- gotcha Handling `'$vector': null` for inserts or updates to collections has seen several fixes across versions 2.0.1 and 2.1.0. While intended to allow setting a vector field to null (removing its vector), users might encounter unexpected behavior if not on the latest patch or if the API behavior shifts.
- deprecated The `alive` property on cursor objects was removed. Instead of `cursor.alive`, you should now check `cursor.state != CursorState.CLOSED`.
Install
-
pip install astrapy
Imports
- AstraDB
from astrapy.db import AstraDB
- Collection
from astrapy.db import Collection
Quickstart
import os
from astrapy.db import AstraDB
# Get credentials from environment variables
ASTRA_DB_API_ENDPOINT = os.environ.get('ASTRA_DB_API_ENDPOINT', '')
ASTRA_DB_APPLICATION_TOKEN = os.environ.get('ASTRA_DB_APPLICATION_TOKEN', '')
ASTRA_DB_KEYSPACE = os.environ.get('ASTRA_DB_KEYSPACE', None) # Optional
if not ASTRA_DB_API_ENDPOINT or not ASTRA_DB_APPLICATION_TOKEN:
raise ValueError(
"Please set ASTRA_DB_API_ENDPOINT and ASTRA_DB_APPLICATION_TOKEN "
"environment variables."
)
# Initialize Astra DB client
astradb = AstraDB(
api_endpoint=ASTRA_DB_API_ENDPOINT,
token=ASTRA_DB_APPLICATION_TOKEN,
namespace=ASTRA_DB_KEYSPACE,
)
print(f"Connected to Astra DB: {astradb.api_endpoint}")
# Get or create a collection
collection = astradb.collection("my_test_collection")
# Insert a document
document_id = "doc1"
inserted_document = collection.insert_one({"_id": document_id, "item": "apple", "price": 1.0})
print(f"Inserted document with ID: {inserted_document['inserted_id']}")
# Find a document
found_document = collection.find_one({"_id": document_id})
print(f"Found document: {found_document}")
# Clean up (optional)
# collection.delete_one({"_id": document_id})
# print(f"Deleted document with ID: {document_id}")