Adding custom requests (#67)

Author: apetenchea

arangoasync/database.py

@@ -2861,6 +2861,21 @@ def response_handler(resp: Response) -> Any:
 
         return await self._executor.execute(request, response_handler)
 
+    async def request(self, request: Request) -> Result[Response]:
+        """Execute a custom request.
+
+        Args:
+            request (Request): Request object to be executed.
+
+        Returns:
+            Response: Response object containing the result of the request.
+        """
+
+        def response_handler(resp: Response) -> Response:
+            return resp
+
+        return await self._executor.execute(request, response_handler)
+
 
 class StandardDatabase(Database):
     """Standard database API wrapper.

docs/database.rst

@@ -14,6 +14,7 @@ information.
 
     from arangoasync import ArangoClient
     from arangoasync.auth import Auth
+    from arangoasync.request import Method, Request
 
     # Initialize the client for ArangoDB.
     async with ArangoClient(hosts="http://localhost:8529") as client:
@@ -60,4 +61,10 @@ information.
         # Delete the database. Note that the new users will remain.
         await sys_db.delete_database("test")
 
+        # Example of a custom request
+        request = Request(
+            method=Method.POST, endpoint="/_admin/execute", data="return 1".encode("utf-8")
+        )
+        response = await sys_db.request(request)
+
 See :class:`arangoasync.client.ArangoClient` and :class:`arangoasync.database.StandardDatabase` for API specification.

tests/test_database.py

@@ -1,5 +1,6 @@
 import asyncio
 import datetime
+import json
 
 import pytest
 from packaging import version
@@ -36,6 +37,7 @@
     ServerTimeError,
     ServerVersionError,
 )
+from arangoasync.request import Method, Request
 from arangoasync.typings import CollectionType, KeyOptions, UserInfo
 from tests.helpers import generate_col_name, generate_db_name, generate_username
 
@@ -157,6 +159,13 @@ async def test_database_misc_methods(
         )
         await db.compact()
 
+    # Custom Request
+    request = Request(
+        method=Method.POST, endpoint="/_admin/execute", data="return 1".encode("utf-8")
+    )
+    response = await sys_db.request(request)
+    assert json.loads(response.raw_body) == 1
+
 
 @pytest.mark.asyncio
 async def test_create_drop_database(