diff --git a/arango/database.py b/arango/database.py index e33bd5ae..5c761c5b 100644 --- a/arango/database.py +++ b/arango/database.py @@ -8,7 +8,7 @@ from datetime import datetime from numbers import Number -from typing import Any, List, Optional, Sequence, Union +from typing import Any, List, Optional, Sequence, Union, Dict from warnings import warn from arango.api import ApiGroup @@ -81,6 +81,8 @@ ViewRenameError, ViewReplaceError, ViewUpdateError, + ServerLogSettingError, + ServerLogSettingSetError, ) from arango.executor import ( AsyncApiExecutor, @@ -749,6 +751,52 @@ def response_handler(resp: Response) -> Json: return self._execute(request, response_handler) + def log_settings(self) -> Result[Json]: + """Return the structured log settings. + + :return: Current log settings. False values are not returned. + :rtype: dict + """ + request = Request(method="get", endpoint="/_admin/log/structured") + + def response_handler(resp: Response) -> Json: + if not resp.is_success: + raise ServerLogSettingError(resp, request) + result: Json = resp.body + return result + + return self._execute(request, response_handler) + + def set_log_settings(self, **kwargs: Dict[str, Any]) -> Result[Json]: + """Set the structured log settings. + + This method takes arbitrary keyword arguments where the keys are the + structured log parameters and the values are true or false, for either + enabling or disabling the parameters. + + .. code-block:: python + + arango.set_log_settings( + database=True, + url=True, + username=False, + ) + + :param kwargs: Structured log parameters. + :type kwargs: Dict[str, Any] + :return: New log settings. False values are not returned. + :rtype: dict + """ + request = Request(method="put", endpoint="/_admin/log/structured", data=kwargs) + + def response_handler(resp: Response) -> Json: + if not resp.is_success: + raise ServerLogSettingSetError(resp, request) + result: Json = resp.body + return result + + return self._execute(request, response_handler) + def log_levels(self, server_id: Optional[str] = None) -> Result[Json]: """Return current logging levels. @@ -775,7 +823,7 @@ def response_handler(resp: Response) -> Json: return self._execute(request, response_handler) def set_log_levels( - self, server_id: Optional[str] = None, **kwargs: str + self, server_id: Optional[str] = None, **kwargs: Dict[str, Any] ) -> Result[Json]: """Set the logging levels. @@ -797,6 +845,8 @@ def set_log_levels( JWT authentication whereas Coordinators also support authentication using usernames and passwords. :type server_id: str | None + :param kwargs: Logging levels. + :type kwargs: Dict[str, Any] :return: New logging levels. :rtype: dict """ diff --git a/arango/exceptions.py b/arango/exceptions.py index fb11f8d5..525d112f 100644 --- a/arango/exceptions.py +++ b/arango/exceptions.py @@ -666,10 +666,18 @@ class ServerLogLevelError(ArangoServerError): """Failed to retrieve server log levels.""" +class ServerLogSettingError(ArangoServerError): + """Failed to retrieve server log settings.""" + + class ServerLogLevelSetError(ArangoServerError): """Failed to set server log levels.""" +class ServerLogSettingSetError(ArangoServerError): + """Failed to set server log settings.""" + + class ServerReloadRoutingError(ArangoServerError): """Failed to reload routing details.""" diff --git a/tests/test_database.py b/tests/test_database.py index da6307a4..13d8ac3e 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -253,6 +253,22 @@ def test_database_misc_methods(sys_db, db, bad_db, cluster): with assert_raises(ServerLogLevelSetError): bad_db.set_log_levels(**new_levels) + # Test Log Settings + result_1 = sys_db.set_log_settings(database=True, url=True, username=True) + result_2 = sys_db.log_settings() + assert isinstance(result_1, dict) + assert "database" in result_1 + assert "url" in result_1 + assert "username" in result_1 + assert result_1 == result_2 + + result_1 = sys_db.set_log_settings(database=True, username=False) + result_2 = sys_db.log_settings() + assert "database" in result_1 + assert "url" in result_1 + assert "username" not in result_1 + assert result_1 == result_2 + # Test get storage engine engine = db.engine() assert engine["name"] in ["rocksdb"]