### Create Foxx Service with File

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Installs a new Foxx service using a local javascript file or zip bundle. Supports configuration, dependency management, and control over development, setup, and legacy modes.

```APIDOC
## POST /_api/foxx

### Description
Installs a new Foxx service using a javascript file or zip bundle.

### Method
POST

### Endpoint
/_api/foxx

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").
- **development** (bool | None) - Optional - Enable development mode.
- **setup** (bool | None) - Optional - Run service setup script.
- **legacy** (bool | None) - Optional - Install the service in 2.8 legacy compatibility mode.

#### Request Body
- **filename** (str) - Required - Full path to the javascript file or zip bundle.
- **config** (dict | None) - Optional - Configuration values.
- **dependencies** (dict | None) - Optional - Dependency settings.

### Response
#### Success Response (200)
- **metadata** (dict) - Service metadata.

### Error Handling
- **aioarango.exceptions.FoxxServiceCreateError**: If install fails.
```

--------------------------------

### Example User Entry for Database Creation

Source: https://aioarango.readthedocs.io/en/latest/specs.html

This is an example entry for the 'users' parameter when creating a new database. It specifies username, password, and active status.

```python
{
    'username': 'john',
    'password': 'password',
    'active': True,
    'extra': {'Department': 'IT'}
}

```

--------------------------------

### Create New Foxx Service

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Installs a new Foxx service using a JSON definition. Requires the service mount path and source (URL or file path). Optional parameters include configuration, dependencies, development mode, setup, and legacy mode.

```python
await foxx.create_service("/my/mount/path", "/path/to/source.zip", config={"key": "value"}, dependencies={"dep_name": "dep_version"}, development=True, setup=True, legacy=False)
```

--------------------------------

### Start Replication Applier

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/replication.html

Starts the replication applier. Optionally, you can specify the last log tick from which to begin applying replication.

```APIDOC
## PUT /_api/replication/applier-start

### Description
Starts the replication applier.

### Method
PUT

### Endpoint
/_api/replication/applier-start

### Parameters
#### Query Parameters
- **from** (str) - Optional - The remote last log tick value from which to start applying replication.

### Response
#### Success Response (200)
- **dict** - Applier state and details.

#### Response Example
```json
{
  "state": "running",
  "lastAppliedContinuousTick": "1234567890",
  "totalRequests": 1001,
  "totalFailedConnects": 0,
  "totalApplyTime": 120.6
}
```
```

--------------------------------

### Create Foxx Service

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Installs a new Foxx service from a source. This method allows for configuration, dependency management, and control over development, setup, and legacy modes.

```APIDOC
## POST /_api/foxx

### Description
Installs a new Foxx service from a source.

### Method
POST

### Endpoint
/_api/foxx

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").
- **development** (bool | None) - Optional - Enable development mode.
- **setup** (bool | None) - Optional - Run service setup script.
- **legacy** (bool | None) - Optional - Install the service in 2.8 legacy compatibility mode.

#### Request Body
- **source** (str) - Required - Fully qualified URL or absolute path on the server file system. Must be accessible by the server, or by all servers if in a cluster.
- **config** (dict | None) - Optional - Configuration values.
- **dependencies** (dict | None) - Optional - Dependency settings.

### Response
#### Success Response (200)
- **metadata** (dict) - Service metadata.

### Error Handling
- **aioarango.exceptions.FoxxServiceCreateError**: If install fails.
```

--------------------------------

### List Services

Source: https://aioarango.readthedocs.io/en/latest/foxx.html

Lists all installed Foxx services.

```APIDOC
## List Services

### Description
Lists all installed Foxx services on the ArangoDB instance.

### Method
`foxx.services()`

### Parameters
None

### Response
#### Success Response (200)
- A list of dictionaries, where each dictionary represents a Foxx service.

### Response Example
```json
[
  {
    "mount": "/example_service",
    "name": "example_service",
    "version": "1.0.0",
    "development": false
  }
]
```
```

--------------------------------

### Install and Run Flake8 for PEP8 Compliance

Source: https://aioarango.readthedocs.io/en/latest/contributing.html

Install flake8 to check for PEP8 compliance. Clone the repository and run flake8 in the project directory.

```bash
~$ pip install flake8
~$ git clone https://github.com/mirrorrim/aioarango.git
~$ cd aioarango
~$ flake8

```

--------------------------------

### List Installed Services

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Retrieves a list of all installed Foxx services on the server. System services can be optionally excluded.

```APIDOC
## GET /api/foxx

### Description
List installed services.

### Method
GET

### Endpoint
/_api/foxx

### Parameters
#### Query Parameters
- **excludeSystem** (bool) - Optional - If set to True, system services are excluded.

### Response
#### Success Response (200)
- **services** (list) - List of installed service metadata.

#### Response Example
```json
[
  {
    "mount": "/_system/my-service",
    "name": "my-service",
    "version": "1.0.0",
    "development": false
  }
]
```
```

--------------------------------

### Create Edge Definition Example

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/graph.html

Provides an example structure for creating a new edge definition, specifying the edge collection and the connected vertex collections.

```python
{
                'edge_collection': 'edge_collection_name',
                'from_vertex_collections': ['from_vertex_collection_name'],
                'to_vertex_collections': ['to_vertex_collection_name']
            }
```

--------------------------------

### List Installed Foxx Services

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Retrieves a list of all installed Foxx services. Set exclude_system to True to omit system services.

```python
await foxx.services(exclude_system=True)
```

--------------------------------

### Get Foxx Service Readme

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Retrieves the readme file for a specified Foxx service.

```APIDOC
## GET /_api/foxx/readme

### Description
Returns the readme file for the specified Foxx service.

### Method
GET

### Endpoint
/_api/foxx/readme

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").

### Response
#### Success Response (200)
- **readme** (str) - The content of the service's readme file.

### Response Example
{
  "readme": "# My Service\n\nThis is a sample service."
}
```

--------------------------------

### Get aioarango Client Version

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/client.html

Retrieve the installed version of the aioarango client library.

```python
version = client.version

```

--------------------------------

### Build Documentation with Sphinx

Source: https://aioarango.readthedocs.io/en/latest/contributing.html

Install Sphinx and the Read the Docs theme, then build the HTML documentation locally. Open the generated 'index.html' file in your browser.

```bash
~$ pip install sphinx sphinx_rtd_theme
~$ git clone https://github.com/mirrorrim/aioarango.git
~$ cd aioarango/docs
~$ sphinx-build . build  # Open build/index.html in a browser

```

--------------------------------

### Initialize and Connect to ArangoDB

Source: https://aioarango.readthedocs.io/en/latest/graph.html

Sets up the ArangoDB client and connects to a specific database. Ensure you have the ArangoClient imported and provide valid connection details.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')
```

--------------------------------

### Get AQL Query Tracking Properties

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/aql.html

Retrieves the current configuration for AQL query tracking, including whether it's enabled and thresholds. This is essential for performance monitoring setup.

```python
    async def tracking(self) -> Result[Json]:
        """Return AQL query tracking properties.

        :return: AQL query tracking properties.
        :rtype: dict
        :raise aioarango.exceptions.AQLQueryTrackingGetError: If retrieval fails.
        """
        request = Request(method="get", endpoint="/_api/query/properties")

        def response_handler(resp: Response) -> Json:
            if not resp.is_success:
                raise AQLQueryTrackingGetError(resp, request)
            return format_aql_tracking(resp.body)

        return await self._execute(request, response_handler)
```

--------------------------------

### Replace Foxx Service

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Replaces an existing Foxx service with a new one from a specified file (JavaScript or zip bundle). It supports options for teardown, setup, legacy mode, and force installation.

```APIDOC
## PUT /_api/foxx/service

### Description
Replace a service using a javascript file or zip bundle.

### Method
PUT

### Endpoint
/_api/foxx/service

### Parameters
#### Path Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").

#### Query Parameters
- **teardown** (bool | None) - Optional - Run service teardown script.
- **setup** (bool | None) - Optional - Run service setup script.
- **legacy** (bool | None) - Optional - Replace the service in 2.8 legacy compatibility mode.
- **force** (bool | None) - Optional - Force install if no service is found.

#### Request Body
- **filename** (str) - Required - Full path to the javascript file or zip bundle.
- **config** (dict | None) - Optional - Configuration values.
- **dependencies** (dict | None) - Optional - Dependency settings.

### Response
#### Success Response (200)
- **metadata** (dict) - Replaced service metadata.

#### Response Example
```json
{
  "metadata": {
    "name": "my-service",
    "version": "1.0.0",
    "mount": "/_admin/my-service"
  }
}
```

### Errors
- **aioarango.exceptions.FoxxServiceReplaceError**: If replace fails.
```

--------------------------------

### Manage Foxx Services with Files using aioarango

Source: https://aioarango.readthedocs.io/en/latest/foxx.html

This snippet demonstrates how to create, update, and replace Foxx services by directly providing zip or Javascript files. It includes the necessary steps for initializing the client, connecting to the database, and managing services via file uploads.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "_system" database as root user.
db = await client.db('_system', username='root', password='passwd')

# Get the Foxx API wrapper.
foxx = db.foxx

# Define the test mount point.
service_mount = '/test_mount'

# Create a service by providing a file directly.
await foxx.create_service_with_file(
    mount=service_mount,
    filename='/home/user/service.zip',
    development=True,
    setup=True,
    legacy=True
)

# Update (upgrade) a service by providing a file directly.
await foxx.update_service_with_file(
    mount=service_mount,
    filename='/home/user/service.zip',
    teardown=False,
    setup=True,
    legacy=True,
    force=False
)

# Replace a service by providing a file directly.
await foxx.replace_service_with_file(
    mount=service_mount,
    filename='/home/user/service.zip',
    teardown=False,
    setup=True,
    legacy=True,
    force=False
)

# Delete a service.
await foxx.delete_service(service_mount)

```

--------------------------------

### Configure and Manage Replication Applier

Source: https://aioarango.readthedocs.io/en/latest/replication.html

This snippet shows how to configure, start, stop, and retrieve the state of the ArangoDB replication applier. It includes setting connection parameters, timeouts, and collection restrictions. Ensure the endpoint and authentication details are correct for your setup.

```python
# Get the replication applier configuration.
await replication.applier_config()

# Update the replication applier configuration.
result = await replication.set_applier_config(
    endpoint='http://127.0.0.1:8529',
    database='test',
    username='root',
    password='passwd',
    max_connect_retries=120,
    connect_timeout=15,
    request_timeout=615,
    chunk_size=0,
    auto_start=True,
    adaptive_polling=False,
    include_system=True,
    auto_resync=True,
    auto_resync_retries=3,
    initial_sync_max_wait_time=405,
    connection_retry_wait_time=25,
    idle_min_wait_time=2,
    idle_max_wait_time=3,
    require_from_present=False,
    verbose=True,
    restrict_type='include',
    restrict_collections=['students']
)

# Get the replication applier state.
await replication.applier_state()

# Start the replication applier.
await replication.start_applier()

# Stop the replication applier.
await replication.stop_applier()
```

--------------------------------

### Manage Users and Permissions with aioarango

Source: https://aioarango.readthedocs.io/en/latest/user.html

This comprehensive example demonstrates various operations for managing users and permissions using the aioarango client. It covers user lifecycle management and permission control for databases and collections. Ensure you are connected to the `_system` database as an admin user to perform these actions.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "_system" database as root user.
sys_db = await client.db('_system', username='root', password='passwd')

# List all users.
await sys_db.users()

# Create a new user.
await sys_db.create_user(
    username='johndoe@gmail.com',
    password='first_password',
    active=True,
    extra={'team': 'backend', 'title': 'engineer'}
)

# Check if a user exists.
await sys_db.has_user('johndoe@gmail.com')

# Retrieve details of a user.
await sys_db.user('johndoe@gmail.com')

# Update an existing user.
await sys_db.update_user(
    username='johndoe@gmail.com',
    password='second_password',
    active=True,
    extra={'team': 'frontend', 'title': 'engineer'}
)

# Replace an existing user.
await sys_db.replace_user(
    username='johndoe@gmail.com',
    password='third_password',
    active=True,
    extra={'team': 'frontend', 'title': 'architect'}
)

# Retrieve user permissions for all databases and collections.
await sys_db.permissions('johndoe@gmail.com')

# Retrieve user permission for "test" database.
await sys_db.permission(
    username='johndoe@gmail.com',
    database='test'
)

# Retrieve user permission for "students" collection in "test" database.
await sys_db.permission(
    username='johndoe@gmail.com',
    database='test',
    collection='students'
)

# Update user permission for "test" database.
await sys_db.update_permission(
    username='johndoe@gmail.com',
    permission='rw',
    database='test'
)

# Update user permission for "students" collection in "test" database.
await sys_db.update_permission(
    username='johndoe@gmail.com',
    permission='ro',
    database='test',
    collection='students'
)

# Reset user permission for "test" database.
await sys_db.reset_permission(
    username='johndoe@gmail.com',
    database='test'
)

# Reset user permission for "students" collection in "test" database.
await sys_db.reset_permission(
    username='johndoe@gmail.com',
    database='test',
    collection='students'
)

```

--------------------------------

### Manage Standard Collections in aioarango

Source: https://aioarango.readthedocs.io/en/latest/collection.html

Demonstrates how to initialize the client, connect to a database, and perform various operations on standard collections such as listing, creating, retrieving properties, and deleting.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# List all collections in the database.
await db.collections()

# Create a new collection named "students" if it does not exist.
# This returns an API wrapper for "students" collection.
if await db.has_collection('students'):
    students = db.collection('students')
else:
    students = await db.create_collection('students')

# Retrieve collection properties.
students.name
students.db_name
await students.properties()
await students.revision()
await students.statistics()
await students.checksum()
await students.count()

# Perform various operations.
await students.load()
await students.unload()
await students.truncate()
await students.configure()

# Delete the collection.
await db.delete_collection('students')

```

--------------------------------

### Perform Async API Executions with aioarango

Source: https://aioarango.readthedocs.io/en/latest/async.html

This example demonstrates initiating asynchronous API executions, managing AsyncJob objects, and handling potential errors. It covers inserting documents, executing AQL queries, checking job status, retrieving results, and managing job lifecycles like cancellation and clearing.

```python
import time
import asyncio

from aioarango import (
    ArangoClient,
    AQLQueryExecuteError,
    AsyncJobCancelError,
    AsyncJobClearError
)

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# Begin async execution. This returns an instance of AsyncDatabase, a
# database-level API wrapper tailored specifically for async execution.
async_db = db.begin_async_execution(return_result=True)

# Child wrappers are also tailored for async execution.
async_aql = async_db.aql
async_col = async_db.collection('students')

# API execution context is always set to "async".
assert async_db.context == 'async'
assert async_aql.context == 'async'
assert async_col.context == 'async'

# On API execution, AsyncJob objects are returned instead of results.
job1 = await async_col.insert({'_key': 'Neal'})
job2 = await async_col.insert({'_key': 'Lily'})
job3 = await async_aql.execute('RETURN 100000')
job4 = await async_aql.execute('INVALID QUERY')  # Fails due to syntax error.

# Retrieve the status of each async job.
for job in [job1, job2, job3, job4]:
    # Job status can be "pending", "done" or "cancelled".
    assert await job.status() in {'pending', 'done', 'cancelled'}

    # Let's wait until the jobs are finished.
    while await job.status() != 'done':
        asyncio.sleep(0.1)

# Retrieve the results of successful jobs.
metadata = await job1.result()
assert metadata['_id'] == 'students/Neal'

metadata = await job2.result()
assert metadata['_id'] == 'students/Lily'

cursor = await job3.result()
assert await cursor.next() == 100000

# If a job fails, the exception is propagated up during result retrieval.
try:
    await result = job4.result()
except AQLQueryExecuteError as err:
    assert err.http_code == 400
    assert err.error_code == 1501
    assert 'syntax error' in err.message

# Cancel a job. Only pending jobs still in queue may be cancelled.
# Since job3 is done, there is nothing to cancel and an exception is raised.
try:
    await job3.cancel()
except AsyncJobCancelError as err:
    assert err.message.endswith(f'job {job3.id} not found')

# Clear the result of a job from ArangoDB server to free up resources.
# Result of job4 was removed from the server automatically upon retrieval,
# so attempt to clear it raises an exception.
try:
    await job4.clear()
except AsyncJobClearError as err:
    assert err.message.endswith(f'job {job4.id} not found')

# List the IDs of the first 100 async jobs completed.
await db.async_jobs(status='done', count=100)

# List the IDs of the first 100 async jobs still pending.
await db.async_jobs(status='pending', count=100)

# Clear all async jobs still sitting on the server.
await db.clear_async_jobs()

```

--------------------------------

### Install aioarango

Source: https://aioarango.readthedocs.io/en/latest/index.html

Install the aioarango package using pip. Ensure your Python version is 3.7+ and ArangoDB version is 3.7+.

```bash
~$ pip install aioarango

```

--------------------------------

### enable_development

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Puts a Foxx service into development mode. In development mode, the service is reloaded from the file system and its setup script is re-executed on each request.

```APIDOC
## POST /_api/foxx/development

### Description
Puts a Foxx service into development mode.

### Method
POST

### Endpoint
/_api/foxx/development

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").

### Response
#### Success Response (200)
- **service_metadata** (dict) - Service metadata.

### Response Example
```json
{
  "name": "my_service",
  "mount": "/_admin/my_service",
  "development": true
}
```
```

--------------------------------

### Get Server Statistics

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves statistics from the ArangoDB server. Set `description` to `True` to get a more detailed breakdown of the statistics.

```python
# Get server statistics
await db.statistics()

# Get detailed server statistics
await db.statistics(description=True)
```

--------------------------------

### Run Integration Tests with Coverage Report

Source: https://aioarango.readthedocs.io/en/latest/contributing.html

Install necessary packages for coverage reporting and run the integration tests. This provides a code coverage report for the 'kq' module.

```bash
~$ pip install coverage pytest pytest-cov
~$ git clone https://github.com/mirrorrim/aioarango.git
~$ cd aioarango
~$ pytest --complete --host=127.0.0.1 --port=8529 --passwd=passwd --cov=kq

```

--------------------------------

### Batch Execution with Context Manager

Source: https://aioarango.readthedocs.io/en/latest/batch.html

Demonstrates how to use a context manager for batch API execution, automatically committing the batch upon exiting the context. It shows inserting documents and executing AQL queries within the batch, including handling potential errors.

```APIDOC
## Batch Execution with Context Manager

### Description
This example shows how to initiate batch API execution using an asynchronous context manager. Requests made within the `async with db.begin_batch_execution(...)` block are queued and automatically committed when the block is exited. The `return_result=True` argument ensures that `BatchJob` objects are returned for each operation, allowing for result retrieval and error checking.

### Method
`db.begin_batch_execution(return_result=True)`

### Endpoint
N/A (Client-side operation)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from aioarango import ArangoClient, AQLQueryExecuteError

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# Get the API wrapper for "students" collection.
students = db.collection('students')

# Begin batch execution via context manager.
async with db.begin_batch_execution(return_result=True) as batch_db:

    # Child wrappers are also tailored for batch execution.
    batch_aql = batch_db.aql
    batch_col = batch_db.collection('students')

    # API execution context is always set to "batch".
    assert batch_db.context == 'batch'
    assert batch_aql.context == 'batch'
    assert batch_col.context == 'batch'

    # BatchJob objects are returned instead of results.
    job1 = await batch_col.insert({'_key': 'Kris'})
    job2 = await batch_col.insert({'_key': 'Rita'})
    job3 = await batch_aql.execute('RETURN 100000')
    job4 = await batch_aql.execute('INVALID QUERY')  # Fails due to syntax error.

# Upon exiting context, batch is automatically committed.
assert await students.has('Kris', check_rev=False)
assert await students.has('Rita', check_rev=False)

# Retrieve the status of each batch job.
for job in batch_db.queued_jobs():
    assert job.status() in {'pending', 'done'}

# Retrieve the results of successful jobs.
metadata = job1.result()
assert metadata['_id'] == 'students/Kris'

metadata = job2.result()
assert metadata['_id'] == 'students/Rita'

cursor = job3.result()
assert cursor.next() == 100000

# If a job fails, the exception is propagated up during result retrieval.
try:
    result = job4.result()
except AQLQueryExecuteError as err:
    assert err.http_code == 400
    assert err.error_code == 1501
    assert 'syntax error' in err.message
```

### Response
#### Success Response (200)
- **BatchJob objects**: Returned for each operation when `return_result=True`.

#### Response Example
`job1` (BatchJob object for insert)
`job3` (BatchJob object for AQL query)

#### Error Response
- **AQLQueryExecuteError**: Raised when an AQL query fails during `job.result()` retrieval.
```

--------------------------------

### Manage ArangoDB Databases with aioarango

Source: https://aioarango.readthedocs.io/en/latest/database.html

Initialize the client, connect to the _system database, and perform operations like listing, creating, and deleting databases. This snippet also demonstrates creating a database with specific users and updating permissions.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "_system" database as root user.
# This returns an API wrapper for "_system" database.
sys_db = await client.db('_system', username='root', password='passwd')

# List all databases.
await sys_db.databases()

# Create a new database named "test" if it does not exist.
# Only root user has access to it at time of its creation.
if not await sys_db.has_database('test'):
    await sys_db.create_database('test')

# Delete the database.
await sys_db.delete_database('test')

# Create a new database named "test" along with a new set of users.
# Only "jane", "john", "jake" and root user have access to it.
if not await sys_db.has_database('test'):
    await sys_db.create_database(
        name='test',
        users=[
            {'username': 'jane', 'password': 'foo', 'active': True},
            {'username': 'john', 'password': 'bar', 'active': True},
            {'username': 'jake', 'password': 'baz', 'active': True},
        ],
    )

# Connect to the new "test" database as user "jane".
db = await client.db('test', username='jane', password='foo')

# Make sure that user "jane" has read and write permissions.
await sys_db.update_permission(username='jane', permission='rw', database='test')

# Retrieve various database and server information.
db.name
db.username
await db.version()
await db.status()
await db.details()
await db.collections()
await db.graphs()
await db.engine()

# Delete the database. Note that the new users will remain.
await sys_db.delete_database('test')
```

--------------------------------

### Graph Edge Definition Example

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

An example of how to structure the edge definitions for creating a graph. This specifies the edge collection and the vertex collections it connects.

```python
{
    'edge_collection': 'teach',
    'from_vertex_collections': ['teachers'],
    'to_vertex_collections': ['lectures']
}
```

--------------------------------

### Connecting to Multiple Coordinators

Source: https://aioarango.readthedocs.io/en/latest/cluster.html

Demonstrates how to initialize the ArangoClient to connect to multiple ArangoDB coordinators using either a list of host strings or a comma-separated string.

```APIDOC
## Connecting to Multiple Coordinators

To connect to multiple ArangoDB coordinators, you must provide either a list of host strings or a comma-separated string during client initialization.

### Example:
```python
from aioarango import ArangoClient

# Single host
client = ArangoClient(hosts='http://localhost:8529')

# Multiple hosts (option 1: list)
client = ArangoClient(hosts=['http://host1:8529', 'http://host2:8529'])

# Multiple hosts (option 2: comma-separated string)
client = ArangoClient(hosts='http://host1:8529,http://host2:8529')
```

By default, a httpx.AsyncClient instance is created per coordinator. HTTP requests to a host are sent using only its corresponding session. For more information on how to override this behaviour, see HTTP Clients.
```

--------------------------------

### Edge Definition Body Example

Source: https://aioarango.readthedocs.io/en/latest/graph.html

An example JSON structure representing an edge definition, specifying the edge collection and the connected vertex collections.

```json
{
    'edge_collection': 'teach',
    'from_vertex_collections': ['teachers'],
    'to_vertex_collections': ['lectures']
}

```

--------------------------------

### Remove documents by example

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/collection.html

Removes documents from a collection based on a provided example filter. Supports optional synchronization and limiting the number of deleted documents.

```python
        data: Json = {"collection": self.name, "example": filters}
        if sync is not None:
            data["waitForSync"] = sync
        if limit is not None and limit != 0:
            data["limit"] = limit

        request = Request(
            method="put",
            endpoint="/_api/simple/remove-by-example",
            data=data,
            write=self.name,
        )

        def response_handler(resp: Response) -> int:
            if resp.is_success:
                result: int = resp.body["deleted"]
                return result
            raise DocumentDeleteError(resp, request)

        return await self._execute(request, response_handler)
```

--------------------------------

### Manage Graphs with aioarango

Source: https://aioarango.readthedocs.io/en/latest/graph.html

Demonstrates initializing the client, connecting to a database, listing graphs, creating a new graph, retrieving its properties, and deleting the graph.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# List existing graphs in the database.
await db.graphs()

# Create a new graph named "school" if it does not already exist.
# This returns an API wrapper for "school" graph.
if await db.has_graph('school'):
    school = db.graph('school')
else:
    school = await db.create_graph('school')

# Retrieve various graph properties.
school.name
school.db_name
await school.vertex_collections()
await school.edge_definitions()

# Delete the graph.
await db.delete_graph('school')

```

--------------------------------

### Execute AQL Query with Options

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/aql.html

Constructs and executes an AQL query with a comprehensive set of options. Use this when fine-grained control over query execution is needed.

```python
        if full_count is not None:
            options["fullCount"] = full_count
        if max_plans is not None:
            options["maxNumberOfPlans"] = max_plans
        if optimizer_rules is not None:
            options["optimizer"] = {"rules": optimizer_rules}
        if fail_on_warning is not None:
            options["failOnWarning"] = fail_on_warning
        if profile is not None:
            options["profile"] = profile
        if max_transaction_size is not None:
            options["maxTransactionSize"] = max_transaction_size
        if max_warning_count is not None:
            options["maxWarningCount"] = max_warning_count
        if intermediate_commit_count is not None:
            options["intermediateCommitCount"] = intermediate_commit_count
        if intermediate_commit_size is not None:
            options["intermediateCommitSize"] = intermediate_commit_size
        if satellite_sync_wait is not None:
            options["satelliteSyncWait"] = satellite_sync_wait
        if stream is not None:
            options["stream"] = stream
        if skip_inaccessible_cols is not None:
            options["skipInaccessibleCollections"] = skip_inaccessible_cols
        if max_runtime is not None:
            options["maxRuntime"] = max_runtime

        if options:
            data["options"] = options
        data.update(options)

        request = Request(method="post", endpoint="/_api/cursor", data=data)

        def response_handler(resp: Response) -> Cursor:
            if not resp.is_success:
                raise AQLQueryExecuteError(resp, request)
            return Cursor(self._conn, resp.body)

        return await self._execute(request, response_handler)
```

--------------------------------

### Get Revision

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves the revision of the collection.

```APIDOC
## revision() -> Optional[Union[str, aioarango.job.AsyncJob[str], aioarango.job.BatchJob[str]]]

### Description
Return collection revision.

### Method
async

### Returns
- Optional[Union[str, aioarango.job.AsyncJob[str], aioarango.job.BatchJob[str]]] - Collection revision.

### Raises
- aioarango.exceptions.CollectionRevisionError - If retrieval fails.
```

--------------------------------

### Create Database

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Creates a new database with the specified name and optional configurations.

```APIDOC
## POST /_api/database

### Description
Creates a new database with the specified name and optional configurations.

### Method
POST

### Endpoint
/_api/database

### Parameters
#### Request Body
- **name** (str) - Required - The name of the database to create.
- **users** (list) - Optional - A list of users with access to the new database.
  Each user is a dictionary with fields "username", "password", "active", and "extra".
- **replication_factor** (int | str) - Optional - Default replication factor for collections.
- **write_concern** (int) - Optional - Default write concern for collections.
- **sharding** (str) - Optional - Sharding method for new collections.

### Request Example
```json
{
  "name": "my_new_database",
  "users": [
    {
      "username": "john",
      "password": "password",
      "active": true,
      "extra": {"Department": "IT"}
    }
  ],
  "options": {
    "replicationFactor": "satellite",
    "writeConcern": 2,
    "sharding": "flexible"
  }
}
```

### Response
#### Success Response (200)
- **result** (bool) - True if the database was created successfully.
```

--------------------------------

### Pregel Job Management with aioarango

Source: https://aioarango.readthedocs.io/en/latest/pregel.html

Demonstrates initializing the ArangoDB client, connecting to a database, and performing Pregel operations including creating a job, retrieving job details, and deleting a job. Ensure the 'school' graph and 'pagerank' algorithm are available.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# Get the Pregel API wrapper.
pregel = db.pregel

# Start a new Pregel job in "school" graph.
job_id = await db.pregel.create_job(
    graph='school',
    algorithm='pagerank',
    store=False,
    max_gss=100,
    thread_count=1,
    async_mode=False,
    result_field='result',
    algorithm_params={'threshold': 0.000001}
)

# Retrieve details of a Pregel job by ID.
job = await pregel.job(job_id)

# Delete a Pregel job by ID.
await pregel.delete_job(job_id)

```

--------------------------------

### Get Statistics

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves the statistics for the collection.

```APIDOC
## statistics() -> Optional[Union[Dict[str, Any], aioarango.job.AsyncJob[Dict[str, Any]], aioarango.job.BatchJob[Dict[str, Any]]]]

### Description
Return collection statistics.

### Method
async

### Returns
- Optional[Union[Dict[str, Any], aioarango.job.AsyncJob[Dict[str, Any]], aioarango.job.BatchJob[Dict[str, Any]]]] - Collection statistics.

### Raises
- aioarango.exceptions.CollectionStatisticsError - If retrieval fails.
```

--------------------------------

### Create Service

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Installs a new Foxx service using a JSON definition. The source can be a URL or a file path.

```APIDOC
## POST /api/foxx/service

### Description
Install a new service using JSON definition.

### Method
POST

### Endpoint
/_api/foxx/service

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").
- **source** (str) - Required - Fully qualified URL or absolute path on the server file system. Must be accessible by the server, or by all servers if in a cluster.
- **config** (dict) - Optional - Configuration values.
- **dependencies** (dict) - Optional - Dependency settings.
- **development** (bool) - Optional - Enable development mode.
- **setup** (bool) - Optional - Run setup script.
- **legacy** (bool) - Optional - Use legacy installation.

### Response
#### Success Response (200)
- **service** (dict) - Metadata of the newly created service.

#### Response Example
```json
{
  "mount": "/_system/my-new-service",
  "name": "my-new-service",
  "version": "1.0.0",
  "development": true
}
```
```

--------------------------------

### Get Properties

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves the properties of the collection.

```APIDOC
## properties() -> Optional[Union[Dict[str, Any], aioarango.job.AsyncJob[Dict[str, Any]], aioarango.job.BatchJob[Dict[str, Any]]]]

### Description
Returns collection properties.

### Method
async

### Returns
- Optional[Union[Dict[str, Any], aioarango.job.AsyncJob[Dict[str, Any]], aioarango.job.BatchJob[Dict[str, Any]]]] - Collection properties.

### Raises
- aioarango.exceptions.CollectionPropertiesError - If retrieval fails.
```

--------------------------------

### Begin Transaction

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Initiates a database transaction. Specify collections for read, write, or exclusive access, and configure transaction parameters like sync, lock timeout, and max size.

```python
await db.begin_transaction(
    read=["users", "posts"],
    write=["comments"],
    sync=True,
    lock_timeout=5
)
```

--------------------------------

### Get All Views

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves a list of all views and their summaries.

```APIDOC
## GET /views

### Description
Return list of views and their summaries.

### Method
GET

### Endpoint
/views

### Response
#### Success Response (200)
- **views** (List[dict]) - List of views.

#### Response Example
{
  "example": "[{\"name\": \"view1\"}, {\"name\": \"view2\"}]"
}
```

--------------------------------

### Configure and Use ArangoDB WAL API

Source: https://aioarango.readthedocs.io/en/latest/wal.html

This snippet demonstrates how to initialize the ArangoDB client, connect to the _system database, and interact with the WAL API for configuration, transaction listing, flushing, and retrieving tick values. Ensure you have admin privileges and are connected to the _system database.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "_system" database as root user.
sys_db = await client.db('_system', username='root', password='passwd')

# Get the WAL API wrapper.
wal = sys_db.wal

# Configure WAL properties.
await wal.configure(
    historic_logs=15,
    oversized_ops=False,
    log_size=30000000,
    reserve_logs=5,
    throttle_limit=0,
    throttle_wait=16000
)

# Retrieve WAL properties.
await wal.properties()

# List WAL transactions.
await wal.transactions()

# Flush WAL with garbage collection.
await wal.flush(garbage_collect=True)

# Get the available ranges of tick values.
await wal.tick_ranges()

# Get the last available tick value.
await wal.last_tick()

# Get recent WAL operations.
await wal.tail()

```

--------------------------------

### dependencies

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/foxx.html

Retrieves the dependency settings for a Foxx service.

```APIDOC
## GET /_api/foxx/dependencies

### Description
Retrieves the dependency settings for a Foxx service.

### Method
GET

### Endpoint
/_api/foxx/dependencies

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path (e.g "/_admin/aardvark").

### Response
#### Success Response (200)
- **dependencies** (dict) - Dependency settings.

### Response Example
```json
{
  "some_dependency": {
    "name": "some_service",
    "mount": "/_admin/some_service"
  }
}
```
```

--------------------------------

### Get All User Details

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves details for all users.

```APIDOC
## GET /users

### Description
Return all user details.

### Method
GET

### Endpoint
/users

### Response
#### Success Response (200)
- **users** (List[dict]) - List of user details.

#### Response Example
{
  "example": "[{\"username\": \"testuser1\"}, {\"username\": \"testuser2\"}]"
}
```

--------------------------------

### Get All Users

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Retrieves details for all users in the database.

```APIDOC
## GET /_api/user

### Description
Return all user details.

### Method
GET

### Endpoint
/_api/user

### Response
#### Success Response (200)
- **result** (list[dict]) - List of user details, each containing 'username', 'active', and 'extra' fields.
```

--------------------------------

### Cursor Profile

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/cursor.html

Get the performance profile of the cursor.

```APIDOC
## profile()

### Description
Return cursor performance profile.

### Returns
- **Optional[Json]**: Cursor performance profile (dict).
```

--------------------------------

### Begin Async Execution

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Initiates asynchronous execution mode for database operations. Use this to get an API wrapper that returns AsyncJob instances for retrieving results later.

```python
db.begin_async_execution()
```

--------------------------------

### Initialize ArangoDB Client with Multiple Hosts

Source: https://aioarango.readthedocs.io/en/latest/cluster.html

Provide a list of host strings or a comma-separated string to connect to multiple ArangoDB coordinators. By default, a httpx.AsyncClient instance is created per coordinator.

```python
from aioarango import ArangoClient

# Single host
client = ArangoClient(hosts='http://localhost:8529')

# Multiple hosts (option 1: list)
client = ArangoClient(hosts=['http://host1:8529', 'http://host2:8529'])

# Multiple hosts (option 2: comma-separated string)
client = ArangoClient(hosts='http://host1:8529,http://host2:8529')
```

--------------------------------

### Get View Details

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Retrieves the details of a specific view.

```APIDOC
## _async _view(_name : str_)

### Description
Return view details.

### Parameters
#### Path Parameters
- **name** (str) - Required - The name of the view to retrieve details for.

### Returns
View details.

Return type
    
dict

### Raises
    
**aioarango.exceptions.ViewGetError** – If retrieval fails.
```

--------------------------------

### Get User

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Retrieves details for a specific user by username.

```APIDOC
## GET /_api/user/{username}

### Description
Return user details.

### Method
GET

### Endpoint
/api/user/{username}

### Parameters
#### Path Parameters
- **username** (str) - Required - The username of the user to retrieve.

### Response
#### Success Response (200)
- **username** (str) - The username of the user.
- **active** (bool) - Whether the user is active.
- **extra** (dict) - Additional data for the user.

#### Response Example
{
  "username": "john_doe",
  "active": true,
  "extra": {}
}
```

--------------------------------

### Manage ArangoDB Collection Indexes with aioarango

Source: https://aioarango.readthedocs.io/en/latest/indexes.html

Demonstrates initializing the client, connecting to a database, creating a collection, and then adding various types of indexes including hash, fulltext, skiplist, geo, persistent, and TTL indexes. It also shows how to name an index and delete it.

```python
from aioarango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient()

# Connect to "test" database as root user.
db = await client.db('test', username='root', password='passwd')

# Create a new collection named "cities".
cities = await db.create_collection('cities')

# List the indexes in the collection.
await cities.indexes()

# Add a new hash index on document fields "continent" and "country".
index = await cities.add_hash_index(fields=['continent', 'country'], unique=True)

# Add new fulltext indexes on fields "continent" and "country".
index = await cities.add_fulltext_index(fields=['continent'])
index = await cities.add_fulltext_index(fields=['country'])

# Add a new skiplist index on field 'population'.
index = await cities.add_skiplist_index(fields=['population'], sparse=False)

# Add a new geo-spatial index on field 'coordinates'.
index = await cities.add_geo_index(fields=['coordinates'])

# Add a new persistent index on field 'currency'.
index = await cities.add_persistent_index(fields=['currency'], sparse=True)

# Add a new TTL (time-to-live) index on field 'currency'.
index = await cities.add_ttl_index(fields=['ttl'], expiry_time=200)

# Indexes may be added with a name that can be referred to in AQL queries.
index = await cities.add_hash_index(fields=['country'], name='my_hash_index')

# Delete the last index from the collection.
await cities.delete_index(index['id'])
```

--------------------------------

### Begin Batch Execution

Source: https://aioarango.readthedocs.io/en/latest/_modules/aioarango/database.html

Initiates batch execution mode for database operations. Use this to get an API wrapper that returns BatchJob instances for results upon commit.

```python
db.begin_batch_execution()
```

--------------------------------

### Get Document Count

Source: https://aioarango.readthedocs.io/en/latest/specs.html

Returns the total number of documents in the collection.

```APIDOC
## count() -> Optional[Union[int, aioarango.job.AsyncJob[int], aioarango.job.BatchJob[int]]]

### Description
Return the total document count.

### Method
async

### Returns
- Optional[Union[int, aioarango.job.AsyncJob[int], aioarango.job.BatchJob[int]]] - Total document count.
```