python-arango-async

https://python-arango-async.readthedocs.io/en/latest/
Admin
python-arango-async is an async Python driver for ArangoDB that enables developers to interact with...
Tokens:263,482
Snippets:1,338
Trust Score:9.6
Update:6 months ago
Show doc for...
Tokens:
Raw
### GET /_admin/options

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/database.html

Retrieves the current server startup options.

```APIDOC
## GET /_admin/options

### Description
Returns the currently-set server options.

### Method
GET

### Endpoint
/_admin/options

### Response
#### Success Response (200)
- **result** (dict) - Server options.
```

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

### GET /foxx/services

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Retrieves a list of installed Foxx services.

```APIDOC
## GET /foxx/services

### Description
List installed services.

### Parameters
#### Query Parameters
- **exclude_system** (bool) - Optional - Exclude system services.

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

### Errors
- **FoxxServiceListError** - If retrieval fails.
```

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

### Install New Service

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Installs a new Foxx service at the specified mount path. Supports various payload formats and installation options.

```APIDOC
## POST /_api/foxx/service

### Description
Installs a new Foxx service at the given mount path. The service payload can be a JSON string, a file-like object, or a multipart form.

### Method
POST

### Endpoint
`/_api/foxx/service`

### Query Parameters
- **mount** (string) - Required - The mount path where the service should be installed.
- **development** (boolean) - Optional - If true, installs the service in development mode.
- **setup** (boolean) - Optional - If true, runs the service setup script after installation.
- **legacy** (boolean) - Optional - If true, installs the service in legacy mode.

### Request Body
- **service** (any) - Required - The service payload. Can be a JSON string, a file-like object, or a multipart form.

### Response
#### Success Response (200)
- **metadata** (object) - An object containing the metadata of the newly installed service.

### Response Example
```json
{
  "mount": "/_system/new-service",
  "name": "new-service",
  "version": "1.0.0",
  "development": false,
  "legacy": false,
  "main": "index.js"
}
```
```

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

### Insert Vertices and Edges for Graph Traversal Example

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/graph.rst.txt

Sets up the graph structure by inserting vertices and edges, preparing for graph traversal queries. This example uses specific vertex and edge collection wrappers.

```python
from arangoasync import ArangoClient
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

    # Get the API wrapper for graph "school".
    if await db.has_graph("school"):
        school = db.graph("school")
    else:
        school = await db.create_graph("school")

    # Create vertex collections "lectures" and "teachers" if they do not exist.
    if not await school.has_vertex_collection("lectures"):
        await school.create_vertex_collection("lectures")
    if not await school.has_vertex_collection("teachers"):
        await school.create_vertex_collection("teachers")

    # Create the edge collection "teach".
    if not await school.has_edge_definition("teach"):
        await school.create_edge_definition(
            edge_collection="teach",
            from_vertex_collections=["teachers"],
            to_vertex_collections=["lectures"]
        )

    # Get API wrappers for "from" and "to" vertex collections.
    teachers = school.vertex_collection("teachers")
    lectures = school.vertex_collection("lectures")

    # Get the API wrapper for the edge collection.
    teach = school.edge_collection("teach")

    # Insert vertices into the graph.
    await teachers.insert({"_key": "jon", "name": "Professor jon"})
    await lectures.insert({"_key": "CSC101", "name": "Introduction to CS"})
    await lectures.insert({"_key": "MAT223", "name": "Linear Algebra"})
    await lectures.insert({"_key": "STA201", "name": "Statistics"})

    # Insert edges into the graph.
    await teach.insert({"_from": "teachers/jon", "_to": "lectures/CSC101"})
    await teach.insert({"_from": "teachers/jon", "_to": "lectures/STA201"})
    await teach.insert({"_from": "teachers/jon", "_to": "lectures/MAT223"})
```

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

### Install python-arango-async

Source: https://python-arango-async.readthedocs.io/en/latest/index.html

Install the python-arango-async package using pip. Ensure you have Python 3.10+ and ArangoDB 3.11+.

```bash
~$ pip install python-arango-async --upgrade

```

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

### GET /_admin/support-info

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/database.html

Retrieves deployment information for support purposes.

```APIDOC
## GET /_admin/support-info

### Description
Retrieves deployment information for support purposes. Note: This API can only be accessed from inside the _system database.

### Method
GET

### Endpoint
/_admin/support-info

### Response
#### Success Response (200)
- **result** (dict) - Deployment information.
```

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

### GET /dependencies

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Retrieve the dependency settings for a specific Foxx service.

```APIDOC
## GET /dependencies

### Description
Return service dependencies.

### Parameters
#### Query Parameters
- **mount** (str) - Required - Service mount path.

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

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

### GET /_admin/options-description

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/database.html

Retrieves a description of all available server startup options.

```APIDOC
## GET /_admin/options-description

### Description
Return a description of all available server options.

### Method
GET

### Endpoint
/_admin/options-description

### Response
#### Success Response (200)
- **(object)**: Server options description.

### Response Example
```json
{
  "example": "response body"
}
```
```

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

### GET /edges

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Retrieves edges starting or ending at the specified vertex.

```APIDOC
## GET /edges

### Description
Return the edges starting or ending at the specified vertex.

### Method
GET

### Parameters
#### Query Parameters
- **collection** (str) - Required - Name of the edge collection to return edges from.
- **vertex** (str|dict) - Required - Document ID, key or body.
- **direction** (str) - Optional - Direction of the edges to return. Selects in or out direction for edges.
- **allow_dirty_read** (bool) - Optional - Allow reads from followers in a cluster.

### Response
#### Success Response (200)
- **result** (dict) - List of edges and statistics.
```

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

### GET /_api/edges/{collection}

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/collection.html

Retrieves a list of edges starting or ending at a specified vertex.

```APIDOC
## GET /_api/edges/{collection}

### Description
Return the edges starting or ending at the specified vertex.

### Method
GET

### Endpoint
/_api/edges/{collection}

### Parameters
#### Query Parameters
- **vertex** (str | dict) - Required - Document ID, key or body.
- **direction** (str) - Optional - Direction of the edges to return ('in' or 'out'). If not set, any edges are returned.

### Response
#### Success Response (200)
- **edges** (list) - List of edges.
- **stats** (dict) - Statistics related to the edge retrieval.
```

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

### Deploy Foxx services from local files

Source: https://python-arango-async.readthedocs.io/en/latest/foxx.html

Shows how to deploy services using multipart/form-data or raw binary data when the server cannot access the source URL directly.

```python
import aiofiles
import aiohttp
import json
from arangoasync import ArangoClient
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

    # Define the test mount points.
    mount_point = "/test_mount"

    # Create the service using multipart/form-data.
    service = aiohttp.FormData()
    service.add_field(
        "source",
        open("./tests/static/service.zip", "rb"),
        filename="service.zip",
        content_type="application/zip",
    )
    service.add_field("configuration", json.dumps({}))
    service.add_field("dependencies", json.dumps({}))
    service_info = await db.foxx.create_service(
        mount=mount_point, service=service, headers={"content-type": "multipart/form-data"}
    )

    # Replace the service using raw data.
    async with aiofiles.open("./tests/static/service.zip", mode="rb") as f:
        service = await f.read()
        service_info = await db.foxx.replace_service(
            mount=mount_point, service=service, headers={"content-type": "application/zip"}
        )

    # Delete the service.
    await db.foxx.delete_service(mount_point)
```

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

### Initialize KeyOptions

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/typings.html

Initializes KeyOptions with specified parameters. If data is provided, other parameters are ignored.

```python
def __init__(
        self,
        allow_user_keys: bool = True,
        generator_type: str = "traditional",
        increment: Optional[int] = None,
        offset: Optional[int] = None,
        data: Optional[Json] = None,
    ) -> None:
        if data is None:
            data: Json = {  # type: ignore[no-redef]
                "allowUserKeys": allow_user_keys,
                "type": generator_type,
            }
            if increment is not None:
                data["increment"] = increment  # type: ignore[index]
            if offset is not None:
                data["offset"] = offset  # type: ignore[index]
        super().__init__(cast(Json, data))
```

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

### GET /_api/collection/{collection}/shards

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/collection.html

Returns collection shards and properties, available only in a cluster setup.

```APIDOC
## GET /_api/collection/{collection}/shards

### Description
Return collection shards and properties. Available only in a cluster setup.

### Method
GET

### Endpoint
/_api/collection/{collection}/shards

### Parameters
#### Path Parameters
- **collection** (string) - Required - The name of the collection.

#### Query Parameters
- **details** (boolean) - Optional - If set to true, include responsible servers for these shards.

### Response
#### Success Response (200)
- **shards** (object) - Collection shards information.
```

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

### Get Service Metadata

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Retrieves metadata for a specific Foxx service installed at a given mount path.

```APIDOC
## GET /_api/foxx/service

### Description
Retrieves metadata for a specific Foxx service.

### Method
GET

### Endpoint
`/_api/foxx/service`

### Query Parameters
- **mount** (string) - Required - The mount path of the service (e.g., "/_admin/aardvark").

### Response
#### Success Response (200)
- **metadata** (object) - An object containing the service's metadata.

### Response Example
```json
{
  "mount": "/_admin/aardvark",
  "name": "aardvark",
  "version": "1.2.3",
  "development": true,
  "legacy": false,
  "main": "index.js",
  "scripts": {
    "main": "index.js"
  },
  "dependencies": {},
  "configuration": {}
}
```
```

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

### Execute AQL Queries with Python

Source: https://python-arango-async.readthedocs.io/en/latest/aql.html

Demonstrates how to initialize the ArangoDB client, connect to a database, and execute AQL queries. Includes examples for retrieving query plans, validating queries, inserting data, iterating through results, and managing running/slow queries. Ensure the ArangoDB server is running and accessible.

```python
from arangoasync import ArangoClient, AQLQueryKillError
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

    # Insert some test documents into "students" collection.
    await students.insert_many([
        {"_key": "Abby", "age": 22},
        {"_key": "John", "age": 18},
        {"_key": "Mary", "age": 21}
    ])

    # Get the AQL API wrapper.
    aql = db.aql

    # Retrieve the execution plan without running the query.
    plan = await aql.explain("FOR doc IN students RETURN doc")

    # Validate the query without executing it.
    validate = await aql.validate("FOR doc IN students RETURN doc")

    # Execute the query
    cursor = await db.aql.execute(
      "FOR doc IN students FILTER doc.age < @value RETURN doc",
      bind_vars={"value": 19}
    )

    # Iterate through the result cursor
    student_keys = []
    async for doc in cursor:
        student_keys.append(doc)

    # List currently running queries.
    queries = await aql.queries()

    # List any slow queries.
    slow_queries = await aql.slow_queries()

    # Clear slow AQL queries if any.
    await aql.clear_slow_queries()

    # Retrieve AQL query tracking properties.
    await aql.tracking()

    # Configure AQL query tracking properties.
    await aql.set_tracking(
        max_slow_queries=10,
        track_bind_vars=True,
        track_slow_queries=True
    )

    # Kill a running query (this should fail due to invalid ID).
    try:
        await aql.kill("some_query_id")
    except AQLQueryKillError as err:
        assert err.http_code == 404

```

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

### Initialize and Use ArangoClient with Context Manager

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/overview.rst.txt

Demonstrates connecting to ArangoDB, creating a database and collection, adding an index, inserting documents, and executing an AQL query using an asynchronous context manager.

```python
from arangoasync import ArangoClient
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "_system" database as root user.
    sys_db = await client.db("_system", auth=auth)

    # Create a new database named "test".
    await sys_db.create_database("test")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

    # Add a persistent index to the collection.
    await students.add_index(type="persistent", fields=["name"], options={"unique": True})

    # Insert new documents into the collection.
    await students.insert({"name": "jane", "age": 39})
    await students.insert({"name": "josh", "age": 18})
    await students.insert({"name": "judy", "age": 21})

    # Execute an AQL query and iterate through the result cursor.
    cursor = await db.aql.execute("FOR doc IN students RETURN doc")
    async with cursor:
        student_names = []
        async for doc in cursor:
            student_names.append(doc["name"])
```

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

### Get Collection Shards

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/collection.html

Retrieves collection shards and their properties. Only available in a cluster setup. Optionally include details about responsible servers.

```python
async def shards(self, details: Optional[bool] = None) -> Result[Json]:
        """Return collection shards and properties.

        Available only in a cluster setup.

        Args:
            details (bool | None): If set to `True`, include responsible
                servers for these shards.

        Returns:
            dict: Collection shards.

        Raises:
            CollectionShardsError: If retrieval fails.

        References:
            - `get-the-shard-ids-of-a-collection <https://docs.arango.ai/arangodb/stable/develop/http-api/collections/#get-the-shard-ids-of-a-collection>`__
        """  # noqa: E501
        params: Params = {}
        if details is not None:
            params["details"] = details

        request = Request(
            method=Method.GET,
            endpoint=f"/_api/collection/{self.name}/shards",
            params=params,
        )

        def response_handler(resp: Response) -> Json:
            if not resp.is_success:
                raise CollectionShardsError(resp, request)
            return cast(Json, self.deserializer.loads(resp.raw_body)["shards"])

        return await self._executor.execute(request, response_handler)
```

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

### Execute and Manage AQL Queries

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/aql.rst.txt

Demonstrates how to execute queries, validate syntax, retrieve execution plans, and manage query tracking and termination.

```python
from arangoasync import ArangoClient, AQLQueryKillError
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

    # Insert some test documents into "students" collection.
    await students.insert_many([
        {"_key": "Abby", "age": 22},
        {"_key": "John", "age": 18},
        {"_key": "Mary", "age": 21}
    ])

    # Get the AQL API wrapper.
    aql = db.aql

    # Retrieve the execution plan without running the query.
    plan = await aql.explain("FOR doc IN students RETURN doc")

    # Validate the query without executing it.
    validate = await aql.validate("FOR doc IN students RETURN doc")

    # Execute the query
    cursor = await db.aql.execute(
      "FOR doc IN students FILTER doc.age < @value RETURN doc",
      bind_vars={"value": 19}
    )

    # Iterate through the result cursor
    student_keys = []
    async for doc in cursor:
        student_keys.append(doc)

    # List currently running queries.
    queries = await aql.queries()

    # List any slow queries.
    slow_queries = await aql.slow_queries()

    # Clear slow AQL queries if any.
    await aql.clear_slow_queries()

    # Retrieve AQL query tracking properties.
    await aql.tracking()

    # Configure AQL query tracking properties.
    await aql.set_tracking(
        max_slow_queries=10,
        track_bind_vars=True,
        track_slow_queries=True
    )

    # Kill a running query (this should fail due to invalid ID).
    try:
        await aql.kill("some_query_id")
    except AQLQueryKillError as err:
        assert err.http_code == 404
```

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

### Graph Traversal AQL Query

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/graph.rst.txt

An example AQL query to perform a graph traversal, starting from a specific vertex and exploring edges up to a certain depth.

```python
# AQL to perform a graph traversal.
# Traverse 1 to 3 hops from the vertex "teachers/jon",
query = """

```

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

### Get server options

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/database.html

Returns the currently configured startup options for the ArangoDB server.

```python
request = Request(method=Method.GET, endpoint="/_admin/options")

        def response_handler(resp: Response) -> Json:
            if not resp.is_success:
                raise ServerCurrentOptionsGetError(resp, request)
            result: Json = self.deserializer.loads(resp.raw_body)
            return result
```

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

### CollectionInfo JSON Example

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/typings.html

Example of JSON structure representing collection information in ArangoDB.

```json
{
    "id" : "151",
    "name" : "animals",
    "status" : 3,
    "type" : 2,
    "isSystem" : false,
}
```

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

### Manage Foxx services via URL source

Source: https://python-arango-async.readthedocs.io/en/latest/foxx.html

Demonstrates standard lifecycle operations including listing, creating, updating, and deleting services using a source URL.

```python
from arangoasync import ArangoClient
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

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

    # List services.
    await foxx.services()

    # Create a service using a source file.
    # In this case, the server must have access to the URL.
    service = {
        "source": "/tests/static/service.zip",
        "configuration": {},
        "dependencies": {},
    }
    await foxx.create_service(
        mount=service_mount,
        service=service,
        development=True,
        setup=True,
        legacy=True
    )

    # Update (upgrade) a service.
    await db.foxx.update_service(
        mount=service_mount,
        service=service,
        teardown=True,
        setup=True,
        legacy=False
    )

    # Replace (overwrite) a service.
    await db.foxx.replace_service(
        mount=service_mount,
        service=service,
        teardown=True,
        setup=True,
        legacy=True,
        force=False
    )

    # Get service details.
    await foxx.service(service_mount)

    # Manage service configuration.
    await foxx.config(service_mount)
    await foxx.update_config(service_mount, options={})
    await foxx.replace_config(service_mount, options={})

    # Manage service dependencies.
    await foxx.dependencies(service_mount)
    await foxx.update_dependencies(service_mount, options={})
    await foxx.replace_dependencies(service_mount, options={})

    # Toggle development mode for a service.
    await foxx.enable_development(service_mount)
    await foxx.disable_development(service_mount)

    # Other miscellaneous functions.
    await foxx.readme(service_mount)
    await foxx.swagger(service_mount)
    await foxx.download(service_mount)
    await foxx.commit()
    await foxx.scripts(service_mount)
    await foxx.run_script(service_mount, "setup", {})
    await foxx.run_tests(service_mount, reporter="xunit", output_format="xml")

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

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

### List Installed Services

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Retrieves a list of all installed Foxx services. System services can be excluded.

```APIDOC
## GET /_api/foxx

### Description
Lists installed Foxx services. You can optionally exclude system services.

### Method
GET

### Endpoint
`/_api/foxx`

### Query Parameters
- **excludeSystem** (boolean) - Optional - If true, system services are excluded from the list.

### Response
#### Success Response (200)
- **services** (array) - A list of installed Foxx services, each represented as a JSON object.

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

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

### Manage Collection Indexes

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/indexes.rst.txt

Demonstrates initializing the client, creating a collection, and performing various index operations including listing, adding, and deleting.

```python
from arangoasync import ArangoClient
from arangoasync.auth import Auth

# Initialize the client for ArangoDB.
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

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

    # Add a new persistent index on document fields "continent" and "country".
    # Indexes may be added with a name that can be referred to in AQL queries.
    persistent_index = await cities.add_index(
        type="persistent",
        fields=['continent', 'country'],
        options={"unique": True, "name": "continent_country_index"}
    )

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

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

    # Add a new TTL (time-to-live) index on field 'currency'.
    index = await cities.add_index(type="ttl", fields=["currency"], options={"expireAfter": 200})

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

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

### Install python-arango-async via pip

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/index.rst.txt

Use this command to install or upgrade the driver in your Python environment.

```bash
~$ pip install python-arango-async --upgrade
```

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

### Initialize ArangoDB Async Client and Execute AQL

Source: https://python-arango-async.readthedocs.io/en/latest/async.html

Demonstrates establishing a connection, inserting documents, and executing an asynchronous AQL query. Ensure the server is reachable at the specified host and credentials are valid.

```python
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

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

    # Insert some documents into the collection.
    await students.insert_many([{"_key": "Neal"}, {"_key": "Lily"}])

    # Begin async execution.
    async_db = db.begin_async_execution(return_result=True)

    aql = async_db.aql
    job = await aql.execute(
        f"FOR d IN {students.name} SORT d._key RETURN d",
        count=True,
        batch_size=1,
        ttl=1000,
    )
    await job.wait()

    # Iterate through the cursor.
    # Although the request to fetch the cursor is async, its underlying executor is no longer async.
    # Next batches will be fetched in real-time.
    doc_cnt = 0
    cursor = await job.result()
    async with cursor as ctx:
        async for _ in ctx:
            doc_cnt += 1
    assert doc_cnt == 2
```

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

### Initialize Client with Default Compression

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/compression.rst.txt

Create an ArangoClient instance with default compression enabled. Ensure `arangoasync` is installed.

```python
from arangoasync import ArangoClient
from arangoasync.compression import DefaultCompressionManager

client = ArangoClient(
    hosts="http://localhost:8529",
    compression=DefaultCompressionManager(),
)
```

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

### List Installed Foxx Services

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Retrieves a list of installed services, with an optional flag to exclude system services.

```python
async def services(self, exclude_system: Optional[bool] = False) -> Result[Jsons]:
        """List installed services.

        Args:
            exclude_system (bool | None): Exclude system services.

        Returns:
            list: List of installed services.

        Raises:
            FoxxServiceListError: If retrieval fails.

        References:
            - `list-the-installed-services <https://docs.arango.ai/arangodb/stable/develop/http-api/foxx/#list-the-installed-services>`__
        """  # noqa: E501
        params: Params = {}
        if exclude_system is not None:
            params["excludeSystem"] = exclude_system

        request = Request(
            method=Method.GET,
            endpoint="/_api/foxx",
            params=params,
        )

        def response_handler(resp: Response) -> Jsons:
            if not resp.is_success:
                raise FoxxServiceListError(resp, request)
            result: Jsons = self.deserializer.loads_many(resp.raw_body)
            return result

        return await self._executor.execute(request, response_handler)
```

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

### Python AQL Query Options Initialization

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/typings.html

Demonstrates initializing AQL query options in Python. This class allows setting various parameters like maxPlans, optimizer rules, and more.

```python
def __init__(
    self,
    allow_dirty_reads: Optional[bool] = None,
    allow_retry: Optional[bool] = None,
    fail_on_warning: Optional[bool] = None,
    fill_block_cache: Optional[bool] = None,
    full_count: Optional[bool] = None,
    intermediate_commit_count: Optional[int] = None,
    intermediate_commit_size: Optional[int] = None,
    max_dnf_condition_members: Optional[int] = None,
    max_nodes_per_callstack: Optional[int] = None,
    max_number_of_plans: Optional[int] = None,
    max_runtime: Optional[Number] = None,
    max_transaction_size: Optional[int] = None,
    max_warning_count: Optional[int] = None,
    optimizer: Optional[Json] = None,
    profile: Optional[int] = None,
    satellite_sync_wait: Optional[Number] = None,
    skip_inaccessible_collections: Optional[bool] = None,
    spill_over_threshold_memory_usage: Optional[int] = None,
    spill_over_threshold_num_rows: Optional[int] = None,
    stream: Optional[bool] = None,
    use_plan_cache: Optional[bool] = None,
) -> None:
    data: Json = dict()
    if allow_dirty_reads is not None:
        data["allowDirtyReads"] = allow_dirty_reads
    if allow_retry is not None:
        data["allowRetry"] = allow_retry
    if fail_on_warning is not None:
        data["failOnWarning"] = fail_on_warning
    if fill_block_cache is not None:
        data["fillBlockCache"] = fill_block_cache
    if full_count is not None:
        data["fullCount"] = full_count
    if intermediate_commit_count is not None:
        data["intermediateCommitCount"] = intermediate_commit_count
    if intermediate_commit_size is not None:
        data["intermediateCommitSize"] = intermediate_commit_size
    if max_dnf_condition_members is not None:
        data["maxDNFConditionMembers"] = max_dnf_condition_members
    if max_nodes_per_callstack is not None:
        data["maxNodesPerCallstack"] = max_nodes_per_callstack
    if max_number_of_plans is not None:
        data["maxNumberOfPlans"] = max_number_of_plans
```

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

### POST /foxx/services

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Installs a new Foxx service at a specified mount path.

```APIDOC
## POST /foxx/services

### Description
Installs the given new service at the given mount path.

### Request Body
- **mount** (str) - Required - Mount path the service should be installed at.
- **service** (Any) - Required - Service payload (JSON string, file-like object, or multipart form).
- **headers** (dict) - Optional - Request headers.
- **development** (bool) - Optional - Whether to install the service in development mode.
- **setup** (bool) - Optional - Whether to run the service setup script.
- **legacy** (bool) - Optional - Whether to install in legacy mode.

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

### Errors
- **FoxxServiceCreateError** - If installation fails.
```

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

### Get Edge

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/graph.html

Retrieves a specific edge from the graph. This method delegates to the underlying edge collection's `get` method.

```APIDOC
## Get Edge

### Description
Retrieves a specific edge from the graph. This method delegates to the underlying edge collection's `get` method.

### Method
GET (implicitly via underlying collection method)

### Endpoint
(Delegated to edge collection)

### Parameters
#### Path Parameters
None

#### Query Parameters
- **rev** (string) - Optional - If this is set, a document is only returned if it has exactly this revision.
- **if_match** (string) - Optional - The document is returned if it has the same revision as the given ETag.
- **if_none_match** (string) - Optional - The document is returned if it has a different revision than the given ETag.

#### Request Body
- **edge** (string | object) - Required - Document ID, key, or body of the edge. If a body is provided, it must contain the `_id` or `_key` field.

### Response
#### Success Response (200)
- **result** (object | null) - The edge document if found, otherwise `null`.

### Errors
- **DocumentRevisionError**: If the revision is incorrect.
- **DocumentGetError**: If retrieval fails.
```

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

### GET /_api/replication/inventory

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/replication.html

Retrieves an overview of collections and indexes. This endpoint can be used to get a snapshot of the replication state for a specific batch.

```APIDOC
## GET /_api/replication/inventory

### Description
Return an overview of collections and indexes.

### Method
GET

### Endpoint
/_api/replication/inventory

### Parameters
#### Query Parameters
- **batchId** (str) - Required - Batch ID.
- **includeSystem** (bool) - Optional - Include system collections.
- **global** (bool) - Optional - Include all databases (only on "_system").
- **collection** (bool) - Optional - If this parameter is set, the response will be restricted to a single collection (the one specified), and no views will be returned.
- **DBServer** (str) - Optional - On a Coordinator, this request must have a DBserver query parameter

### Response
#### Success Response (200)
- **collections** (list) - A list of collections and their indexes.
- **views** (list) - A list of views.

#### Response Example
```json
{
  "collections": [
    {
      "id": "12345",
      "name": "my_collection",
      "type": 2,
      "status": 2,
      "indexes": [
        {
          "id": "67890",
          "type": "primary",
          "fields": ["_key"],
          "unique": true,
          "sparse": false
        }
      ]
    }
  ],
  "views": []
}
```
```

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

### Create a new user

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Demonstrates creating a user using either a UserInfo object or a dictionary.

```python
await db.create_user(UserInfo(user="john", password="secret"))
await db.create_user({user="john", password="secret"})
```

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

### Create a New Foxx Service

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Installs a new service at the specified mount path with optional configuration flags.

```python
async def create_service(
        self,
        mount: str,
        service: Any,
        headers: Optional[RequestHeaders] = None,
        development: Optional[bool] = None,
        setup: Optional[bool] = None,
        legacy: Optional[bool] = None,
    ) -> Result[Json]:
        """Installs the given new service at the given mount path.

        Args:
            mount (str): Mount path the service should be installed at.
            service (Any): Service payload. Can be a JSON string, a file-like object, or a
                multipart form.
            headers (dict | None): Request headers.
            development (bool | None): Whether to install the service in development mode.
            setup (bool | None): Whether to run the service setup script.
            legacy (bool | None): Whether to install in legacy mode.

        Returns:
            dict: Service metadata.

        Raises:
            FoxxServiceCreateError: If installation fails.

        References:
           - `install-a-new-service-mode <https://docs.arango.ai/arangodb/stable/develop/http-api/foxx/#install-a-new-service>`__
        """  # noqa: E501
        params: Params = dict()
        params["mount"] = mount
        if development is not None:
            params["development"] = development
        if setup is not None:
            params["setup"] = setup
```

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

### PATCH /_api/foxx/service

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/foxx.html

Upgrades a Foxx service to a new version or replaces it with a new one. It can also be used to install a service if it is not already installed.

```APIDOC
## PATCH /_api/foxx/service

### Description
Upgrades a Foxx service to a new version or replaces it with a new one. It can also be used to install a service if it is not already installed.

### Method
PATCH

### Endpoint
/_api/foxx/service

### Parameters
#### Query Parameters
- **mount** (str) - Required - The mount path of the service.
- **teardown** (str) - Optional - Defines how to handle dependencies when upgrading.
- **setup** (str) - Optional - Defines how to handle dependencies when upgrading.
- **legacy** (bool) - Optional - Whether to use legacy behavior.
- **force** (bool) - Optional - Whether to force the upgrade.

#### Request Body
- **service** (dict or str) - Required - The service to upgrade or install. Can be a dictionary with service metadata or a string representing the service code.
- **headers** (dict) - Optional - Custom headers to send with the request.

### Request Example
```json
{
  "service": {
    "name": "my-service",
    "version": "1.0.0"
  },
  "headers": {
    "X-ArangoDB-Token": "your-token"
  }
}
```

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

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

### Errors
- **FoxxServiceUpdateError**: If upgrade fails.
```

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

### Initialize ArangoDB Client and Execute AQL

Source: https://python-arango-async.readthedocs.io/en/latest/helpers.html

Establishes a connection to the ArangoDB server and executes an AQL query with specific performance and optimization properties.

```python
async with ArangoClient(hosts="http://localhost:8529") as client:
    auth = Auth(username="root", password="passwd")

    # Connect to "test" database as root user.
    db = await client.db("test", auth=auth)

    properties = QueryProperties(
        allow_dirty_reads=True,
        allow_retry=False,
        fail_on_warning=True,
        fill_block_cache=False,
        full_count=True,
        intermediate_commit_count=1000,
        intermediate_commit_size=1048576,
        max_dnf_condition_members=10,
        max_nodes_per_callstack=100,
        max_number_of_plans=5,
        max_runtime=60.0,
        max_transaction_size=10485760,
        max_warning_count=10,
        optimizer={"rules": ["-all", "+use-indexes"]},
        profile=1,
        satellite_sync_wait=10.0,
        skip_inaccessible_collections=True,
        spill_over_threshold_memory_usage=10485760,
        spill_over_threshold_num_rows=100000,
        stream=True,
        use_plan_cache=True,
    )

    # The types are fully serializable.
    print(properties)

    await db.aql.execute(
        "FOR doc IN students RETURN doc",
        batch_size=1,
        options=properties,
    )
```

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

### Get Vertex Document

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/graph.html

Retrieves a vertex document from the graph. This method delegates to the `get` method of the corresponding vertex collection. Supports `if_match` and `if_none_match` options.

```APIDOC
## GET /_api/collection/{collection_name}/document/{document_key}

### Description
Retrieves a vertex document from the graph.

### Method
GET

### Endpoint
`/_api/collection/{collection_name}/document/{document_key}`

### Parameters
#### Path Parameters
- **collection_name** (string) - Required - The name of the vertex collection.
- **document_key** (string) - Required - The key or ID of the vertex document.

#### Query Parameters
- **if_match** (string) - Optional - The document is returned, if it has the same revision as the given ETag.
- **if_none_match** (string) - Optional - The document is returned, if it has a different revision than the given ETag.

### Response
#### Success Response (200)
- **vertex_document** (object) - The vertex document.

#### Response Example
```json
{
  "_key": "12345",
  "_id": "users/12345",
  "name": "John Doe",
  "age": 30
}
```
```

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

### Example Server Status Information JSON

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/typings.html

Provides an example of the JSON response when querying server status, including version, PID, license type, and operational mode.

```json
{
  "server" : "arango",
  "version" : "3.12.2",
  "pid" : 244,
  "license" : "enterprise",
  "mode" : "server",
  "operationMode" : "server",
  "foxxApi" : true,
  "host" : "localhost",
  "hostname" : "ebd1509c9185",
  "serverInfo" : {
    "progress" : {
      "phase" : "in wait",
      "feature" : "",
      "recoveryTick" : 0
    },
    "maintenance" : false,
    "role" : "COORDINATOR",
    "writeOpsEnabled" : true,
    "readOnly" : false,

```

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

### Create User (Synchronous python-arango)

Source: https://python-arango-async.readthedocs.io/en/latest/_sources/migration.rst.txt

Example of creating a new user using the synchronous python-arango driver.

```python
sys_db.create_user(
    username="johndoe@gmail.com",
    password="first_password",
    active=True,
    extra={"team": "backend", "title": "engineer"}
)
```

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

### Create Database (Python)

Source: https://python-arango-async.readthedocs.io/en/latest/_modules/arangoasync/database.html

Use this method to create a new database. It requires execution from the `_system` database and supports optional user configurations and replication settings.

```python
async def create_database(
        self,
        name: str,
        users: Optional[Sequence[Json | UserInfo]] = None,
        replication_factor: Optional[int | str] = None,
        write_concern: Optional[int] = None,
        sharding: Optional[bool] = None,
    ) -> Result[bool]:
        """Create a new database.

        Note:
            This method can only be executed from within the **_system** database.

        Args:
            name (str): Database name.
            users (list | None): Optional list of users with access to the new
                database, where each user is of :class:`User
                <arangoasync.wrapper.UserInfo>` type, or a dictionary with fields
                "username", "password" and "active". If not set, the default user
                **root** will be used to ensure that the new database will be
                accessible after it is created.
            replication_factor (int | str | None): Default replication factor for new
                collections created in this database. Special values include
                “satellite”, which will replicate the collection to every DB-Server
                (Enterprise Edition only), and 1, which disables replication. Used
                for clusters only.
            write_concern (int | None): Default write concern for collections created
                in this database. Determines how many copies of each shard are required
                to be in sync on different DB-Servers. If there are less than these many
                copies in the cluster a shard will refuse to write. Writes to shards with
                enough up-to-date copies will succeed at the same time, however. Value of
                this parameter can not be larger than the value of **replication_factor**.
                Used for clusters only.
            sharding (str | None): Sharding method used for new collections in this
                database. Allowed values are: "", "flexible" and "single". The first
                two are equivalent. Used for clusters only.

        Returns:
            bool: True if the database was created successfully.

        Raises:
            DatabaseCreateError: If creation fails.

        References:
            - `create-a-database <https://docs.arango.ai/arangodb/stable/develop/http-api/databases/#create-a-database>`__
        """  # noqa: E501
        data: Json = {"name": name}

        options: Json = {}
        if replication_factor is not None:
            options["replicationFactor"] = replication_factor
        if write_concern is not None:
            options["writeConcern"] = write_concern
        if sharding is not None:
            options["sharding"] = sharding
        if options:
            data["options"] = options

        if users is not None:
            data["users"] = [
                {
                    "username": user["user"] if "user" in user else user["username"],
                    "passwd": user["password"],
                    "active": user.get("active", True),
                    "extra": user.get("extra", {}),
                }
                for user in users
            ]

        request = Request(
            method=Method.POST,
            endpoint="/_api/database",
            data=self.serializer.dumps(data),
        )

        def response_handler(resp: Response) -> bool:
            if resp.is_success:
                return True
            msg: Optional[str] = None
            if resp.status_code == HTTP_FORBIDDEN:
                msg = "This request can only be executed in the _system database."

```

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

### GET /graph/properties

Source: https://python-arango-async.readthedocs.io/en/latest/specs.html

Retrieve the properties of the graph.

```APIDOC
## GET /graph/properties

### Description
Get the properties of the graph.

### Method
GET

### Response
#### Success Response (200)
- **properties** (GraphProperties) - Properties of the graph.
```

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

### Configure AQL Query Properties

Source: https://python-arango-async.readthedocs.io/en/latest/helpers.html

Example JSON structure for defining query optimizer rules and plan limits.

```json
{
  "maxPlans": 1,
  "optimizer": {
    "rules": [
      "-all",
      "+remove-unnecessary-filters"
    ]
  }
}
```
Tokens:
Raw