Python Driver for ArangoDB
Python-Arango is the official Python driver for ArangoDB, a scalable multi-model database that natively supports documents, graphs, and key-values. It provides a comprehensive API for interacting with ArangoDB, including managing databases, collections, documents, graphs, and executing AQL queries. The library is actively maintained with regular releases, currently at version 8.3.1.
Warnings
- breaking The transaction API was overhauled in `python-arango` version 5.0.0 to align with ArangoDB 3.5+'s new transaction REST API. This removed context managers (e.g., `with db.begin_transaction():`) and changed method signatures. Code using the old transaction patterns will break.
- gotcha For `asyncio` applications, use the dedicated `python-arango-async` library. `python-arango` is a synchronous driver, and using it in an asynchronous context will lead to blocking I/O and potential performance bottlenecks.
- gotcha Instances of `BatchDatabase` and `BatchJob` are stateful and should not be shared across multiple threads. A `BatchDatabase` instance also cannot be reused after its `commit()` method has been called.
Install
-
pip install python-arango --upgrade
Imports
- ArangoClient
from arango import ArangoClient
Quickstart
import os
from arango import ArangoClient
from arango.exceptions import ServerConnectionError, ArangoClientError, CollectionCreateError, DatabaseCreateError
# Configuration from environment variables (replace with your ArangoDB instance details)
ARANGO_HOSTS = os.environ.get('ARANGO_HOSTS', 'http://localhost:8529')
ARANGO_USERNAME = os.environ.get('ARANGO_USERNAME', 'root')
ARANGO_PASSWORD = os.environ.get('ARANGO_PASSWORD', 'your_arangodb_password') # Default root password is often empty or 'root'
TEST_DB_NAME = "my_test_db_reg"
TEST_COLLECTION_NAME = "my_collection_reg"
try:
# Initialize the client for ArangoDB
client = ArangoClient(hosts=ARANGO_HOSTS)
# Connect to "_system" database as root user to manage other databases
sys_db = client.db("_system", username=ARANGO_USERNAME, password=ARANGO_PASSWORD)
# Create a new database if it doesn't exist
if not sys_db.has_database(TEST_DB_NAME):
sys_db.create_database(TEST_DB_NAME)
print(f"Database '{TEST_DB_NAME}' created.")
else:
print(f"Database '{TEST_DB_NAME}' already exists.")
# Connect to the specific database
db = client.db(TEST_DB_NAME, username=ARANGO_USERNAME, password=ARANGO_PASSWORD)
# Create a new collection if it doesn't exist
if not db.has_collection(TEST_COLLECTION_NAME):
collection = db.create_collection(TEST_COLLECTION_NAME)
print(f"Collection '{TEST_COLLECTION_NAME}' created.")
else:
collection = db.collection(TEST_COLLECTION_NAME)
print(f"Collection '{TEST_COLLECTION_NAME}' already exists.")
# Insert new documents into the collection
docs = [
{"name": "Alice", "age": 30},
{"name": "Bob", "age": 24},
{"name": "Charlie", "age": 35}
]
collection.insert_many(docs)
print(f"Inserted {len(docs)} documents.")
# Execute an AQL query and iterate through the result cursor
cursor = db.aql.execute(f"FOR doc IN {TEST_COLLECTION_NAME} FILTER doc.age > 25 RETURN doc.name")
filtered_names = [name for name in cursor]
print(f"Names of people older than 25: {filtered_names}")
except ServerConnectionError as e:
print(f"Failed to connect to ArangoDB server at {ARANGO_HOSTS}: {e}. Ensure ArangoDB is running.")
except ArangoClientError as e:
print(f"ArangoDB Client Error: {e}")
except (CollectionCreateError, DatabaseCreateError) as e:
print(f"Error during ArangoDB resource creation: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")