aql-builder 0.0.16

ArangoDB AQL builder

ArangoDB AQL builder

Python AQL builder for ArangoDB, a scalable multi-model database natively supporting documents, graphs and search.

Why use this?

Instead of writing raw AQL strings:

query = f'FOR u IN users FILTER u.age >= {min_age} RETURN u'

Use a fluent, chainable API:

query = AB.for_('u').in_('users').filter(AB.ref('u.age').gte(min_age)).return_('u').to_aql()

Benefits:

• Composable: Build complex queries by chaining methods
• Type-safe: Automatic conversion of Python values to AQL types
• Readable: Query structure mirrors the AQL syntax
• Maintainable: Easier to modify and extend queries programmatically

Features

• Fluent, chainable API for building AQL queries
• Support for all AQL operations: FOR, FILTER, LET, COLLECT, SORT, LIMIT, RETURN
• DML operations: INSERT, UPDATE, REPLACE, REMOVE, UPSERT
• Graph traversals with configurable depth and direction
• All ArangoDB built-in functions available as methods
• Automatic Python-to-AQL type conversion
• Bind parameter support (@param, @@collection)
• Query formatting: compact, pretty-printed, and debug modes
• Full type hint support with IDE autocomplete for all 200+ AQL functions

Requirements

• Python version 3.10+

Installation

pip install aql-builder --upgrade

Getting Started

Here are simple usage examples:

>>> from aql_builder import AQLBuilder as AB
>>> 
>>> AB.for_('my').in_('mycollection').return_('my._key').to_aql()
'FOR my IN mycollection RETURN my._key'
>>> 
>>> AB.for_('u').in_('users').replace('u').in_('backup').to_aql()
'FOR u IN users REPLACE u IN backup'
>>> 
>>> AB.for_('u').in_('users')
...     .filter(
...         AB.ref('u.active').eq(True)
...         .and_(AB.ref('u.age').gte(35))
...         .and_(AB.ref('u.age').lte(37))
...     )
...     .remove('u').in_('users')
...     .to_aql()
'FOR u IN users FILTER (((u.active == true) && (u.age >= 35)) && (u.age <= 37)) REMOVE u IN users'
>>> 

Working with query parameters:

>>> from aql_builder import AQLBuilder as AB
>>> 
>>> AB.for_('user').in_('@@users').filter(AB.ref('user.age').gte('@min_age')).return_('user._key').to_aql()
'FOR user IN @@users FILTER (user.age >= @min_age) RETURN user._key'
>>> 

Working with graph:

>>> from aql_builder import AQLBuilder as AB
>>> 
>>> AB.for_('v').in_graph(
...         graph='knowsGraph',
...         direction='OUTBOUND',
...         start_vertex=AB.str('users/1'),
...         min_depth=1,
...         max_depth=3
...     )
...     .return_('v._key')
...     .to_aql()
'FOR v IN 1..3 OUTBOUND "users/1" GRAPH "knowsGraph" RETURN v._key'
>>> 

More complex examples:

>>> from aql_builder import AQLBuilder as AB
>>> 
>>> AB.for_('i').in_(AB.range(1, 1000))
...     .insert({
...         'id': AB(100000).add('i'),
...         'age': AB(18).add(AB.FLOOR(AB.RAND().times(25))),
...         'name': AB.CONCAT('test', AB.TO_STRING('i')),
...         'active': False,
...         'gender': (
...             AB.ref('i').mod(2).eq(0).then('"male"').else_('"female"')
...         )
...     }).into('users')
...     .to_aql()
'FOR i IN 1..1000 INSERT
  {"id": (100000 + i),
   "age": (18 + FLOOR((RAND() * 25))),
   "name": CONCAT(test, TO_STRING(i)),
   "active": false,
   "gender": (((i % 2) == 0) ? "male" : "female")}
  INTO users'
>>> 
>>> 
>>> AB.for_('u').in_('users')
...     .filter(AB.ref('u.active').eq(True))
...     .collect({
...         'ageGroup': AB.FLOOR(AB.ref('u.age').div(5)).times(5),
...         'gender': 'u.gender'
...     }).into('group')
...     .sort('ageGroup', 'DESC')
...     .return_({
...         'ageGroup': 'ageGroup',
...         'gender': 'gender'
...     })
...     .to_aql()
'FOR u IN users
  FILTER (u.active == true)
  COLLECT ageGroup = (FLOOR((u.age / 5)) * 5), gender = u.gender INTO group
  SORT ageGroup DESC
  RETURN {"ageGroup": ageGroup, "gender": gender}'
>>>

Query Formatting

The library supports multiple output formats for AQL queries.

Compact Format (Default)

The default format produces a single-line query with normalized whitespace:

>>> from aql_builder import AQLBuilder as AB
>>>
>>> query = (
...     AB.for_('u').in_('users')
...     .filter(AB.ref('u.age').gte(18))
...     .filter(AB.ref('u.active').eq(True))
...     .sort('u.name', 'ASC')
...     .return_('u')
... )
>>>
>>> query.to_aql()
'FOR u IN users FILTER (u.age >= 18) FILTER (u.active == true) SORT u.name ASC RETURN u'
>>>

Pretty Format

Use format='pretty' for multi-line, indented output:

>>> print(query.to_aql(format='pretty'))
FOR u IN users
    FILTER (u.age >= 18)
    FILTER (u.active == true)
    SORT u.name ASC
    RETURN u
>>>
>>> # Convenience method
>>> print(query.to_aql_pretty())
FOR u IN users
    FILTER (u.age >= 18)
    FILTER (u.active == true)
    SORT u.name ASC
    RETURN u
>>>

Custom Indentation

Customize the indentation string (default is 4 spaces):

>>> print(query.to_aql(format='pretty', indent='  '))
FOR u IN users
  FILTER (u.age >= 18)
  FILTER (u.active == true)
  SORT u.name ASC
  RETURN u
>>>
>>> # Using tabs
>>> print(query.to_aql_pretty(indent='\t'))
FOR u IN users
	FILTER (u.age >= 18)
	FILTER (u.active == true)
	SORT u.name ASC
	RETURN u
>>>

Debug Mode

Add class name annotations to understand the query structure:

>>> query.to_aql(debug=True)
'FOR u IN users FILTER (u.age >= 18) FILTER (u.active == true) SORT u.name ASC RETURN u  # [ForExpression], [FilterExpression], [FilterExpression], [SortExpression], [ReturnExpression]'
>>>
>>> # Debug with pretty format
>>> print(query.to_aql(format='pretty', debug=True))
FOR u IN users            [ForExpression]
    FILTER (u.age >= 18)  [FilterExpression]
    FILTER (u.active == true)  [FilterExpression]
    SORT u.name ASC       [SortExpression]
    RETURN u              [ReturnExpression]
>>>
>>> # Convenience method
>>> query.to_aql_debug()
'FOR u IN users FILTER (u.age >= 18) FILTER (u.active == true) SORT u.name ASC RETURN u  # [ForExpression], [FilterExpression], [FilterExpression], [SortExpression], [ReturnExpression]'
>>>

Formatting Raw AQL Strings

You can also format raw AQL strings using the static method:

>>> raw_aql = 'FOR u IN users FILTER u.age >= 18 RETURN u'
>>>
>>> # Pretty format
>>> print(AB.format_aql(raw_aql, format_type='pretty'))
FOR u IN users
    FILTER u.age >= 18
    RETURN u
>>>
>>> # Compact format (normalizes whitespace)
>>> AB.format_aql(raw_aql, format_type='compact')
'FOR u IN users FILTER u.age >= 18 RETURN u'
>>>

Type Hints and IDE Support

The package includes type stubs (PEP 561) for IDE autocomplete and mypy type checking. All 200+ ArangoDB functions have full type hint support.

from aql_builder import AQLBuilder as AB

# IDE autocomplete works for all functions
AB.CONCAT(...)  # Shows function signature
AB.DATE_NOW()   # Shows return type

# Type checking with mypy
def build_query(min_age: int) -> str:
    return AB.for_('u').in_('users').filter(...).return_('u').to_aql()

Subqueries

The library provides built-in support for AQL subqueries with helper methods for common patterns.

IN Subquery

Check if a value exists in subquery results:

>>> from aql_builder import AQLBuilder as AB
>>>
>>> # Find users who have placed orders
>>> query = (
...     AB.for_('u').in_('users')
...     .filter(
...         AB.ref('u._id').in_subquery(
...             AB.for_('o').in_('orders').return_('o.userId')
...         )
...     )
...     .return_('u')
... )
>>> query.to_aql()
'FOR u IN users FILTER (u._id in (FOR o IN orders RETURN o.userId)) RETURN u'
>>>
>>> # Find users who have NOT placed orders
>>> query = (
...     AB.for_('u').in_('users')
...     .filter(
...         AB.ref('u._id').not_in_subquery(
...             AB.for_('o').in_('orders').return_('o.userId')
...         )
...     )
...     .return_('u')
... )
>>> query.to_aql()
'FOR u IN users FILTER (u._id not in (FOR o IN orders RETURN o.userId)) RETURN u'
>>>

Comparison Subqueries

Compare values against scalar subquery results (e.g., threshold from config):

>>> # Find products above threshold price from config
>>> query = (
...     AB.for_('p').in_('products')
...     .filter(
...         AB.ref('p.price').gt_subquery(
...             AB.for_('c').in_('config')
...             .filter(AB.ref('c.key').eq('premium_threshold'))
...             .return_('c.value')
...         )
...     )
...     .return_('p')
... )
>>> query.to_aql()
'FOR p IN products FILTER (p.price > (FOR c IN config FILTER (c.key == "premium_threshold") RETURN c.value)) RETURN p'
>>>

Other comparison methods: eq_subquery(), gte_subquery(), lt_subquery(), lte_subquery()

Aggregate Comparisons

For comparisons with aggregate functions (AVG, SUM, MIN, MAX), use the aggregate function directly:

>>> # Find products more expensive than average
>>> query = (
...     AB.for_('p').in_('products')
...     .filter(
...         AB.ref('p.price').gt(
...             AB.AVG(AB.for_('p2').in_('products').return_('p2.price'))
...         )
...     )
...     .return_('p')
... )
>>> query.to_aql()
'FOR p IN products FILTER (p.price > AVG((FOR p2 IN products RETURN p2.price))) RETURN p'
>>>

EXISTS Pattern

Check if a subquery returns any results:

>>> # Find users with recent activity
>>> query = (
...     AB.for_('u').in_('users')
...     .filter(
...         AB.exists_subquery(
...             AB.for_('a').in_('activities')
...             .filter(AB.ref('a.userId').eq('u._id'))
...             .return_('a')
...         )
...     )
...     .return_('u')
... )
>>> query.to_aql()
'FOR u IN users FILTER (LENGTH((FOR a IN activities FILTER (a.userId == u._id) RETURN a)) > 0) RETURN u'
>>>
>>> # Find users without activity
>>> query = (
...     AB.for_('u').in_('users')
...     .filter(
...         AB.not_exists_subquery(
...             AB.for_('a').in_('activities')
...             .filter(AB.ref('a.userId').eq('u._id'))
...             .return_('a')
...         )
...     )
...     .return_('u')
... )
>>> query.to_aql()
'FOR u IN users FILTER (LENGTH((FOR a IN activities FILTER (a.userId == u._id) RETURN a)) == 0) RETURN u'
>>>

Subquery Helpers

Get the first result or count from a subquery:

>>> # Get first value from subquery
>>> AB.first_from_subquery(
...     AB.for_('t').in_('thresholds').return_('t.value')
... ).to_aql()
'FIRST((FOR t IN thresholds RETURN t.value))'
>>>
>>> # Count subquery results
>>> AB.length_of_subquery(
...     AB.for_('o').in_('orders').return_('o')
... ).to_aql()
'LENGTH((FOR o IN orders RETURN o))'
>>>

License

This project is licensed under the Apache License 2.0 - see the LICENSE.txt file for details.

Acknowledgements

AQL builder is a free software project hosted at https://foss.heptapod.net. Thanks to the support of Clever Cloud, Octobus and the sponsors of the heptapod project.