Basic user management operations

arangoasync/database.py

@@ -17,6 +17,10 @@
     DatabaseDeleteError,
     DatabaseListError,
     ServerStatusError,
+    UserCreateError,
+    UserDeleteError,
+    UserGetError,
+    UserListError,
 )
 from arangoasync.executor import ApiExecutor, DefaultApiExecutor
 from arangoasync.request import Method, Request
@@ -553,6 +557,164 @@ def response_handler(resp: Response) -> bool:
 
         return await self._executor.execute(request, response_handler)
 
+    async def has_user(self, username: str) -> Result[bool]:
+        """Check if a user exists.
+
+        Args:
+            username (str): Username.
+
+        Returns:
+            bool: True if the user exists, False otherwise.
+
+        Raises:
+            UserListError: If the operation fails.
+        """
+        request = Request(method=Method.GET, endpoint=f"/_api/user/{username}")
+
+        def response_handler(resp: Response) -> bool:
+            if resp.is_success:
+                return True
+            if resp.status_code == HTTP_NOT_FOUND:
+                return False
+            raise UserListError(resp, request)
+
+        return await self._executor.execute(request, response_handler)
+
+    async def user(self, username: str) -> Result[UserInfo]:
+        """Fetches data about a user.
+
+        Args:
+            username (str): Username.
+
+        Returns:
+            UserInfo: User details.
+
+        Raises:
+            UserGetError: If the operation fails.
+
+        References:
+            - `get-a-user` <https://docs.arangodb.com/stable/develop/http-api/users/#get-a-user>`__
+        """  # noqa: E501
+        request = Request(method=Method.GET, endpoint=f"/_api/user/{username}")
+
+        def response_handler(resp: Response) -> UserInfo:
+            if not resp.is_success:
+                raise UserGetError(resp, request)
+            body = self.deserializer.loads(resp.raw_body)
+            return UserInfo(
+                user=body["user"],
+                active=cast(bool, body.get("active")),
+                extra=body.get("extra"),
+            )
+
+        return await self._executor.execute(request, response_handler)
+
+    async def users(self) -> Result[Sequence[UserInfo]]:
+        """Fetches data about all users.
+
+        Without the necessary permissions, you might only get data about the
+        current user.
+
+        Returns:
+            list: User information.
+
+        Raises:
+            UserListError: If the operation fails.
+
+        References:
+            - `list-available-users <https://docs.arangodb.com/stable/develop/http-api/users/#list-available-users>`__
+        """  # noqa: E501
+        request = Request(method=Method.GET, endpoint="/_api/user")
+
+        def response_handler(resp: Response) -> Sequence[UserInfo]:
+            if not resp.is_success:
+                raise UserListError(resp, request)
+            body = self.deserializer.loads(resp.raw_body)
+            return [
+                UserInfo(user=u["user"], active=u.get("active"), extra=u.get("extra"))
+                for u in body["result"]
+            ]
+
+        return await self._executor.execute(request, response_handler)
+
+    async def create_user(
+        self,
+        user: UserInfo,
+    ) -> Result[UserInfo]:
+        """Create a new user.
+
+        Args:
+            user (UserInfo): User information.
+
+        Returns:
+            UserInfo: New user details.
+
+        Raises:
+            ValueError: If the username is missing.
+            UserCreateError: If the operation fails.
+
+        Example:
+            .. code-block:: python
+
+                await db.create_user(UserInfo(user="john", password="secret"))
+
+        References:
+            - `create-a-user <https://docs.arangodb.com/stable/develop/http-api/users/#create-a-user>`__
+        """  # noqa: E501
+        if not user.user:
+            raise ValueError("Username is required.")
+
+        data: Json = user.to_dict()
+        request = Request(
+            method=Method.POST,
+            endpoint="/_api/user",
+            data=self.serializer.dumps(data),
+        )
+
+        def response_handler(resp: Response) -> UserInfo:
+            if not resp.is_success:
+                raise UserCreateError(resp, request)
+            body = self.deserializer.loads(resp.raw_body)
+            return UserInfo(
+                user=body["user"],
+                active=cast(bool, body.get("active")),
+                extra=body.get("extra"),
+            )
+
+        return await self._executor.execute(request, response_handler)
+
+    async def delete_user(
+        self,
+        username: str,
+        ignore_missing: bool = False,
+    ) -> Result[bool]:
+        """Delete a user.
+
+        Args:
+            username (str): Username.
+            ignore_missing (bool): Do not raise an exception on missing user.
+
+        Returns:
+            bool: True if the user was deleted successfully, `False` if the user was
+                not found but **ignore_missing** was set to `True`.
+
+        Raises:
+            UserDeleteError: If the operation fails.
+
+        References:
+            - `remove-a-user <https://docs.arangodb.com/stable/develop/http-api/users/#remove-a-user>`__
+        """  # noqa: E501
+        request = Request(method=Method.DELETE, endpoint=f"/_api/user/{username}")
+
+        def response_handler(resp: Response) -> bool:
+            if resp.is_success:
+                return True
+            if resp.status_code == HTTP_NOT_FOUND and ignore_missing:
+                return False
+            raise UserDeleteError(resp, request)
+
+        return await self._executor.execute(request, response_handler)
+
 
 class StandardDatabase(Database):
     """Standard database API wrapper."""

arangoasync/exceptions.py

@@ -141,3 +141,19 @@ class ServerConnectionError(ArangoServerError):
 
 class ServerStatusError(ArangoServerError):
     """Failed to retrieve server status."""
+
+
+class UserCreateError(ArangoServerError):
+    """Failed to create user."""
+
+
+class UserDeleteError(ArangoServerError):
+    """Failed to delete user."""
+
+
+class UserGetError(ArangoServerError):
+    """Failed to retrieve user details."""
+
+
+class UserListError(ArangoServerError):
+    """Failed to retrieve users."""

tests/helpers.py

@@ -17,3 +17,21 @@ def generate_col_name():
         str: Random collection name.
     """
     return f"test_collection_{uuid4().hex}"
+
+
+def generate_username():
+    """Generate and return a random username.
+
+    Returns:
+        str: Random username.
+    """
+    return f"test_user_{uuid4().hex}"
+
+
+def generate_string():
+    """Generate and return a random unique string.
+
+    Returns:
+        str: Random unique string.
+    """
+    return uuid4().hex

tests/test_user.py

@@ -0,0 +1,50 @@
+import pytest
+
+from arangoasync.auth import Auth
+from arangoasync.client import ArangoClient
+from arangoasync.typings import UserInfo
+from tests.helpers import generate_string, generate_username
+
+
+@pytest.mark.asyncio
+async def test_user_management(url, sys_db_name, root, password):
+    auth = Auth(username=root, password=password)
+
+    # TODO also handle exceptions
+    async with ArangoClient(hosts=url) as client:
+        db = await client.db(sys_db_name, auth_method="basic", auth=auth, verify=True)
+
+        # Create a user
+        username = generate_username()
+        password = generate_string()
+        users = await db.users()
+        assert not any(user.user == username for user in users)
+        assert await db.has_user(username) is False
+
+        # Verify user creation
+        new_user = await db.create_user(
+            UserInfo(
+                user=username,
+                password=password,
+                active=True,
+                extra={"foo": "bar"},
+            )
+        )
+        assert new_user.user == username
+        assert new_user.active is True
+        assert new_user.extra == {"foo": "bar"}
+        users = await db.users()
+        assert sum(user.user == username for user in users) == 1
+        assert await db.has_user(username) is True
+        user = await db.user(username)
+        assert user.user == username
+        assert user.active is True
+
+        # Delete the newly created user
+        assert await db.delete_user(username) is True
+        users = await db.users()
+        assert not any(user.user == username for user in users)
+        assert await db.has_user(username) is False
+
+        # Ignore missing user
+        assert await db.delete_user(username, ignore_missing=True) is False