tessl/pypi-python-arango
Python Driver for ArangoDB, a scalable multi-model database that natively supports documents, graphs, and search.
—
Pending
collections.mddocs/
Document & Collection Operations
Complete document lifecycle management and collection operations for ArangoDB. Supports standard document collections, vertex collections for graphs, and edge collections with comprehensive CRUD operations, batch processing, and indexing.
Capabilities
Collection Management
Create, delete, and manage collections with various configuration options and collection types.
class StandardDatabase:
def collection(self, name: str) -> StandardCollection:
"""
Get collection interface.
Parameters:
- name: str, collection name
Returns:
StandardCollection: Collection interface object
"""
def collections(self) -> Result[List[Json]]:
"""
List all collections.
Returns:
Result[List[Json]]: List of collection information dicts
"""
def create_collection(self, name: str, sync=None, system: bool = False,
**kwargs) -> Result[StandardCollection]:
"""
Create a new collection.
Parameters:
- name: str, collection name
- sync: bool, wait for sync to disk
- system: bool, create as system collection
- **kwargs: additional collection options (type, sharding, etc.)
Returns:
Result[StandardCollection]: Created collection object
"""
def delete_collection(self, name: str, ignore_missing: bool = False,
system=None) -> Result[bool]:
"""
Delete a collection.
Parameters:
- name: str, collection name
- ignore_missing: bool, ignore if collection doesn't exist
- system: bool, allow deletion of system collections
Returns:
Result[bool]: True on success
"""
def has_collection(self, name: str) -> Result[bool]:
"""
Check if collection exists.
Parameters:
- name: str, collection name
Returns:
Result[bool]: True if collection exists
"""Document Operations
Core CRUD operations for individual documents with revision control, return options, and synchronization settings.
class StandardCollection:
@property
def name(self) -> str:
"""Collection name."""
@property
def db_name(self) -> str:
"""Database name."""
def insert(self, document: Json, return_new: bool = False, sync=None,
silent: bool = False, overwrite: bool = False, **kwargs) -> Result:
"""
Insert a document.
Parameters:
- document: dict, document data
- return_new: bool, return inserted document
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
- overwrite: bool, overwrite existing document
- **kwargs: additional insert options
Returns:
Result[Json|bool]: Document metadata or boolean on success
"""
def get(self, document, rev=None, check_rev: bool = True) -> Result:
"""
Get a document.
Parameters:
- document: str or dict, document key or dict with _key
- rev: str, expected revision
- check_rev: bool, check revision matches
Returns:
Result[Json]: Document data or None if not found
"""
def update(self, document: Json, check_rev: bool = True, merge: bool = True,
keep_none: bool = True, return_new: bool = False,
return_old: bool = False, sync=None, silent: bool = False) -> Result:
"""
Update a document.
Parameters:
- document: dict, document with _key and fields to update
- check_rev: bool, check current revision
- merge: bool, merge with existing document
- keep_none: bool, keep null values
- return_new: bool, return updated document
- return_old: bool, return original document
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
Returns:
Result[Json|bool]: Document metadata or boolean
"""
def replace(self, document: Json, check_rev: bool = True,
return_new: bool = False, return_old: bool = False,
sync=None, silent: bool = False) -> Result:
"""
Replace a document completely.
Parameters:
- document: dict, complete replacement document with _key
- check_rev: bool, check current revision
- return_new: bool, return new document
- return_old: bool, return original document
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
Returns:
Result[Json|bool]: Document metadata or boolean
"""
def delete(self, document, rev=None, check_rev: bool = True,
ignore_missing: bool = False, return_old: bool = False,
sync=None, silent: bool = False) -> Result:
"""
Delete a document.
Parameters:
- document: str or dict, document key or dict with _key
- rev: str, expected revision
- check_rev: bool, check revision matches
- ignore_missing: bool, ignore if document doesn't exist
- return_old: bool, return deleted document
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
Returns:
Result[Json|bool]: Document metadata or boolean
"""
def has(self, document, rev=None, check_rev: bool = True) -> Result[bool]:
"""
Check if document exists.
Parameters:
- document: str or dict, document key or dict with _key
- rev: str, expected revision
- check_rev: bool, check revision matches
Returns:
Result[bool]: True if document exists
"""Batch Operations
Efficient bulk operations for processing multiple documents in single requests with batch processing optimizations.
def insert_many(self, documents: Sequence[Json], return_new: bool = False,
sync=None, silent: bool = False, overwrite: bool = False,
**kwargs) -> Result:
"""
Insert multiple documents.
Parameters:
- documents: list, list of document dicts
- return_new: bool, return inserted documents
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
- overwrite: bool, overwrite existing documents
- **kwargs: additional options
Returns:
Result[List[Json]|bool]: List of document metadata or boolean
"""
def update_many(self, documents: Sequence[Json], check_rev: bool = True,
merge: bool = True, keep_none: bool = True,
return_new: bool = False, return_old: bool = False,
sync=None, silent: bool = False) -> Result:
"""
Update multiple documents.
Parameters:
- documents: list, list of document dicts with _key
- check_rev: bool, check revisions
- merge: bool, merge with existing documents
- keep_none: bool, keep null values
- return_new: bool, return updated documents
- return_old: bool, return original documents
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
Returns:
Result[List[Json]|bool]: List of document metadata or boolean
"""
def delete_many(self, documents: Sequence, return_old: bool = False,
check_rev: bool = True, sync=None, silent: bool = False) -> Result:
"""
Delete multiple documents.
Parameters:
- documents: list, list of keys or document dicts
- return_old: bool, return deleted documents
- check_rev: bool, check revisions
- sync: bool, wait for sync to disk
- silent: bool, minimal return data
Returns:
Result[List[Json]|bool]: List of document metadata or boolean
"""Index Management
Create and manage various index types to optimize query performance and enforce constraints.
def indexes(self) -> Result[List[Json]]:
"""
List all indexes on collection.
Returns:
Result[List[Json]]: List of index information dicts
"""
def add_hash_index(self, fields: Sequence[str], unique=None, sparse=None,
deduplicate=None, name=None, in_background=None) -> Result[Json]:
"""
Create hash index.
Parameters:
- fields: list, field names to index
- unique: bool, enforce uniqueness
- sparse: bool, exclude null values
- deduplicate: bool, remove duplicate values
- name: str, index name
- in_background: bool, create in background
Returns:
Result[Json]: Index information dict
"""
def add_skiplist_index(self, fields: Sequence[str], unique=None, sparse=None,
deduplicate=None, name=None, in_background=None) -> Result[Json]:
"""
Create skiplist index for range queries.
Parameters:
- fields: list, field names to index
- unique: bool, enforce uniqueness
- sparse: bool, exclude null values
- deduplicate: bool, remove duplicate values
- name: str, index name
- in_background: bool, create in background
Returns:
Result[Json]: Index information dict
"""
def add_geo_index(self, fields: Sequence[str], ordered=None, name=None,
in_background=None) -> Result[Json]:
"""
Create geo-spatial index.
Parameters:
- fields: list, field names for coordinates
- ordered: bool, longitude/latitude order
- name: str, index name
- in_background: bool, create in background
Returns:
Result[Json]: Index information dict
"""
def add_fulltext_index(self, fields: Sequence[str], min_length=None,
name=None, in_background=None) -> Result[Json]:
"""
Create fulltext search index.
Parameters:
- fields: list, text field names
- min_length: int, minimum word length
- name: str, index name
- in_background: bool, create in background
Returns:
Result[Json]: Index information dict
"""
def delete_index(self, index_id: str, ignore_missing: bool = False) -> Result[bool]:
"""
Delete an index.
Parameters:
- index_id: str, index identifier
- ignore_missing: bool, ignore if index doesn't exist
Returns:
Result[bool]: True on success
"""Graph Collections
Specialized collection types for graph operations with vertex and edge specific functionality.
class VertexCollection(Collection):
"""Vertex collection for graph operations."""
def link(self, from_vertex, to_vertex, data=None, sync=None) -> Result[Json]:
"""
Create edge between vertices.
Parameters:
- from_vertex: str or dict, source vertex
- to_vertex: str or dict, target vertex
- data: dict, edge data
- sync: bool, wait for sync to disk
Returns:
Result[Json]: Edge document metadata
"""
class EdgeCollection(Collection):
"""Edge collection for graph operations."""
def edges(self, vertex, direction=None) -> Result[Json]:
"""
Get edges connected to vertex.
Parameters:
- vertex: str or dict, vertex identifier
- direction: str, edge direction ('in', 'out', 'any')
Returns:
Result[Json]: Edges information dict
"""
def link(self, from_vertex, to_vertex, data=None, sync=None) -> Result[Json]:
"""
Create edge between vertices.
Parameters:
- from_vertex: str or dict, source vertex
- to_vertex: str or dict, target vertex
- data: dict, edge data
- sync: bool, wait for sync to disk
Returns:
Result[Json]: Edge document metadata
"""Usage Examples
Basic Collection Operations
from arango import ArangoClient
client = ArangoClient()
db = client.db('example', username='root', password='password')
# Create collection
students = db.create_collection('students')
# Insert documents
student1 = students.insert({'name': 'Alice', 'age': 22, 'major': 'CS'})
student2 = students.insert({'name': 'Bob', 'age': 21, 'major': 'Math'})
print(f"Inserted: {student1['_key']}")
# Get document
alice = students.get(student1['_key'])
print(f"Retrieved: {alice['name']}")
# Update document
students.update({'_key': student1['_key'], 'age': 23})
# Delete document
students.delete(student2['_key'])Batch Operations
# Batch insert
documents = [
{'name': 'Charlie', 'age': 20, 'major': 'Physics'},
{'name': 'Diana', 'age': 22, 'major': 'Biology'},
{'name': 'Eve', 'age': 21, 'major': 'Chemistry'}
]
results = students.insert_many(documents, return_new=True)
for result in results:
print(f"Inserted: {result['new']['name']}")
# Batch update
updates = [
{'_key': 'charlie_key', 'gpa': 3.8},
{'_key': 'diana_key', 'gpa': 3.9}
]
students.update_many(updates)
# Batch delete
keys_to_delete = ['eve_key', 'old_record_key']
students.delete_many(keys_to_delete, ignore_missing=True)Index Management
# Create indexes for performance
students.add_hash_index(['name'], unique=True)
students.add_skiplist_index(['age', 'gpa'])
students.add_fulltext_index(['major', 'description'])
# Geo index for location data
locations = db.create_collection('locations')
locations.add_geo_index(['coordinates'])
# List all indexes
indexes = students.indexes()
for index in indexes:
print(f"Index: {index['type']} on {index['fields']}")Collection Configuration
# Create collection with options
courses = db.create_collection(
'courses',
sync=True, # Synchronous writes
schema={ # Document validation
'rule': {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'credits': {'type': 'number', 'minimum': 1}
},
'required': ['name', 'credits']
},
'level': 'moderate'
}
)
# Check collection existence
if db.has_collection('students'):
print("Students collection exists")
# Get collection info
collections = db.collections()
for col in collections:
print(f"Collection: {col['name']} (type: {col['type']})")Error Handling
from arango import DocumentNotFoundError, UniqueConstraintViolatedError
try:
# Try to get non-existent document
doc = students.get('nonexistent_key')
if doc is None:
print("Document not found")
# Try to insert duplicate
students.insert({'name': 'Alice', 'age': 22}) # Duplicate name
except DocumentNotFoundError:
print("Document does not exist")
except UniqueConstraintViolatedError:
print("Unique constraint violated")Install with Tessl CLI
npx tessl i tessl/pypi-python-arango