### 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 `__ """ # 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 `__ """ # 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 `__ """ # 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 ` 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 `__ """ # 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" ] } } ```