CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-python-arango

Python Driver for ArangoDB, a scalable multi-model database that natively supports documents, graphs, and search.

—

Pending
Overview
Eval results
Files

aql.mddocs/

AQL Query Interface

Advanced Query Language (AQL) interface for executing complex queries, analyzing performance, managing query cache, and working with user-defined functions. AQL is ArangoDB's SQL-like query language optimized for multi-model data.

Capabilities

Query Execution

Execute AQL queries with extensive configuration options, bind parameters, and result cursors for efficient data retrieval.

class AQL:
    def execute(self, query: str, count: bool = False, batch_size=None,
                ttl=None, bind_vars=None, full_count=None, max_plans=None,
                optimizer_rules=None, cache=None, memory_limit: int = 0,
                fail_on_warning=None, profile=None, **kwargs) -> Result[Cursor]:
        """
        Execute AQL query.
        
        Parameters:
        - query: str, AQL query string
        - count: bool, return total count
        - batch_size: int, result batch size
        - ttl: float, query time-to-live in seconds
        - bind_vars: dict, bind parameters for query
        - full_count: bool, return full count even with LIMIT
        - max_plans: int, maximum optimizer plans to generate
        - optimizer_rules: list, optimizer rules to apply/disable
        - cache: bool, use query result cache
        - memory_limit: int, memory limit in bytes
        - fail_on_warning: bool, fail on warnings
        - profile: bool or int, profiling level
        - **kwargs: additional query options
        
        Returns:
        Result[Cursor]: Query result cursor
        """

Query Analysis

Analyze query execution plans, validate query syntax, and optimize query performance.

def explain(self, query: str, all_plans: bool = False, max_plans=None,
            opt_rules=None, bind_vars=None) -> Result:
    """
    Explain query execution plan.
    
    Parameters:
    - query: str, AQL query to explain
    - all_plans: bool, return all possible plans
    - max_plans: int, maximum plans to generate
    - opt_rules: list, optimizer rules to consider
    - bind_vars: dict, bind parameters
    
    Returns:
    Result[Json|List[Json]]: Execution plan(s)
    """
    
def validate(self, query: str) -> Result[Json]:
    """
    Validate query syntax.
    
    Parameters:
    - query: str, AQL query to validate
    
    Returns:
    Result[Json]: Validation result with syntax info
    """

Query Management

Monitor running queries, kill long-running queries, and manage slow query tracking.

def queries(self) -> Result[List[Json]]:
    """
    List currently running queries.
    
    Returns:
    Result[List[Json]]: List of active query information
    """
    
def kill(self, query_id: str) -> Result[bool]:
    """
    Kill a running query.
    
    Parameters:
    - query_id: str, ID of query to terminate
    
    Returns:
    Result[bool]: True on success
    """
    
def slow_queries(self) -> Result[List[Json]]:
    """
    Get slow query log.
    
    Returns:
    Result[List[Json]]: List of slow queries
    """
    
def clear_slow_queries(self) -> Result[bool]:
    """
    Clear slow query log.
    
    Returns:
    Result[bool]: True on success
    """

Query Tracking

Configure query tracking settings to monitor performance and identify slow queries.

def tracking(self) -> Result[Json]:
    """
    Get current query tracking configuration.
    
    Returns:
    Result[Json]: Tracking configuration dict
    """
    
def set_tracking(self, enabled=None, max_slow_queries=None,
                 slow_query_threshold=None, max_query_string_length=None,
                 track_bind_vars=None, track_slow_queries=None) -> Result[Json]:
    """
    Configure query tracking.
    
    Parameters:
    - enabled: bool, enable query tracking
    - max_slow_queries: int, max slow queries to track
    - slow_query_threshold: int, threshold for slow queries (seconds)
    - max_query_string_length: int, max query string length to log
    - track_bind_vars: bool, include bind variables in logs
    - track_slow_queries: bool, track slow queries
    
    Returns:
    Result[Json]: Updated tracking configuration
    """

User-Defined Functions

Create, manage, and execute custom AQL functions to extend query capabilities.

def functions(self) -> Result[List[Json]]:
    """
    List all user-defined functions.
    
    Returns:
    Result[List[Json]]: List of function definitions
    """
    
def create_function(self, name: str, code: str) -> Result[Json]:
    """
    Create or update user-defined function.
    
    Parameters:
    - name: str, function name (namespace::name format)
    - code: str, JavaScript function code
    
    Returns:
    Result[Json]: Function creation result
    """
    
def delete_function(self, name: str, group: bool = False,
                    ignore_missing: bool = False) -> Result:
    """
    Delete user-defined function.
    
    Parameters:
    - name: str, function name or namespace
    - group: bool, delete entire function group/namespace
    - ignore_missing: bool, ignore if function doesn't exist
    
    Returns:
    Result[Json|bool]: Deletion result
    """
    
def query_rules(self) -> Result[List[Json]]:
    """
    Get available optimizer rules.
    
    Returns:
    Result[List[Json]]: List of optimizer rule information
    """

Query Cache

Manage AQL query result caching to improve performance for repeated queries.

class AQLQueryCache:
    def properties(self) -> Result[Json]:
        """
        Get query cache properties.
        
        Returns:
        Result[Json]: Cache configuration dict
        """
        
    def configure(self, mode=None, max_results=None, max_results_size=None,
                  max_entry_size=None, include_system=None) -> Result[Json]:
        """
        Configure query cache.
        
        Parameters:
        - mode: str, cache mode ('off', 'on', 'demand')
        - max_results: int, maximum number of cached results
        - max_results_size: int, maximum cache size in bytes
        - max_entry_size: int, maximum size per cache entry
        - include_system: bool, cache system database queries
        
        Returns:
        Result[Json]: Updated cache configuration
        """
        
    def entries(self) -> Result[List[Json]]:
        """
        Get cache entries.
        
        Returns:
        Result[List[Json]]: List of cached queries
        """
        
    def clear(self) -> Result[bool]:
        """
        Clear query cache.
        
        Returns:
        Result[bool]: True on success
        """

Usage Examples

Basic Query Execution

from arango import ArangoClient

client = ArangoClient()
db = client.db('example', username='root', password='password')

# Simple query
query = "FOR doc IN students RETURN doc"
cursor = db.aql.execute(query)
students = [doc for doc in cursor]

# Query with bind parameters
query = "FOR s IN students FILTER s.age >= @min_age RETURN s"
cursor = db.aql.execute(query, bind_vars={'min_age': 21})
adults = list(cursor)

# Query with options
cursor = db.aql.execute(
    query,
    count=True,           # Include count
    batch_size=100,       # Batch size
    cache=True,           # Use cache
    memory_limit=1000000  # 1MB memory limit
)

print(f"Total count: {cursor.count()}")

Complex Queries

# JOIN-like operation
join_query = """
    FOR student IN students
        FOR course IN courses
            FILTER student.major == course.department
            RETURN {
                student_name: student.name,
                course_name: course.name,
                credits: course.credits
            }
"""

# Aggregation query
agg_query = """
    FOR student IN students
        COLLECT major = student.major 
        AGGREGATE count = LENGTH(1), avg_age = AVERAGE(student.age)
        RETURN {
            major: major,
            student_count: count,
            average_age: avg_age
        }
"""

# Graph traversal
graph_query = """
    FOR vertex, edge, path IN 1..3 OUTBOUND 'students/alice' GRAPH 'university'
        RETURN {
            vertex: vertex,
            path_length: LENGTH(path.vertices)
        }
"""

results = db.aql.execute(agg_query)
for result in results:
    print(f"{result['major']}: {result['student_count']} students")

Query Analysis and Optimization

# Explain query execution plan
query = "FOR s IN students FILTER s.age > 21 SORT s.name RETURN s"
plan = db.aql.explain(query)
print(f"Estimated cost: {plan['plan']['estimatedCost']}")

# Validate query syntax
invalid_query = "FOR s IN students FILTER s.age > RETURN s"
validation = db.aql.validate(invalid_query)
if not validation['error']:
    print("Query is valid")
else:
    print(f"Syntax error: {validation['errorMessage']}")

# Profile query execution
cursor = db.aql.execute(query, profile=2)  # Detailed profiling
profile_data = cursor.profile()
print(f"Execution time: {profile_data['executing']}")

User-Defined Functions

# Create custom function
function_code = """
function(name) {
    return "Hello, " + name + "!";
}
"""

db.aql.create_function('greeting::hello', function_code)

# Use custom function in query
query = "FOR s IN students RETURN greeting::hello(s.name)"
cursor = db.aql.execute(query)
greetings = list(cursor)

# List all functions
functions = db.aql.functions()
for func in functions:
    print(f"Function: {func['name']}")

# Delete function
db.aql.delete_function('greeting::hello')

Query Tracking and Monitoring

# Configure tracking
db.aql.set_tracking(
    enabled=True,
    max_slow_queries=100,
    slow_query_threshold=5,  # 5 seconds
    track_bind_vars=True
)

# Monitor running queries
running = db.aql.queries()
for query_info in running:
    print(f"Query {query_info['id']}: {query_info['query'][:50]}...")

# Check slow queries
slow = db.aql.slow_queries()
for slow_query in slow:
    print(f"Slow query took {slow_query['runTime']}s")

# Kill long-running query
if running:
    db.aql.kill(running[0]['id'])

Query Cache Management

# Configure cache
cache = db.aql.cache
cache.configure(
    mode='demand',        # Cache on demand
    max_results=1000,     # Max 1000 cached queries
    max_results_size=64*1024*1024,  # 64MB total
    max_entry_size=1024*1024        # 1MB per entry
)

# Check cache status
cache_props = cache.properties()
print(f"Cache mode: {cache_props['mode']}")
print(f"Cache entries: {len(cache.entries())}")

# Execute cacheable query
query = "FOR s IN students COLLECT age = s.age RETURN {age, count: LENGTH(1)}"
cursor = db.aql.execute(query, cache=True)
results = list(cursor)

# Clear cache when needed
cache.clear()

Advanced Query Options

# Query with optimizer rules
cursor = db.aql.execute(
    "FOR s IN students FILTER s.age > 21 SORT s.name RETURN s",
    optimizer_rules=[
        '+all',              # Enable all rules
        '-sort-limit'        # Disable sort-limit rule
    ]
)

# Query with custom memory limit
large_query = """
    FOR s1 IN students
        FOR s2 IN students
            FILTER s1._key != s2._key
            RETURN {s1: s1.name, s2: s2.name}
"""

try:
    cursor = db.aql.execute(
        large_query,
        memory_limit=10*1024*1024,  # 10MB limit
        fail_on_warning=True
    )
    results = list(cursor)
except Exception as e:
    print(f"Query failed: {e}")

Install with Tessl CLI

npx tessl i tessl/pypi-python-arango

docs

aql.md

client-database.md

collections.md

errors-types.md

graphs.md

index.md

transactions.md

tile.json