feat: enable retrieval of service logs (via API)

doc/changelog.d/1515.added.md

@@ -0,0 +1 @@
+enable retrieval of service logs (via API)

src/ansys/geometry/core/connection/client.py

@@ -28,6 +28,8 @@
 from typing import Optional
 import warnings
 
+from ansys.geometry.core.errors import protect_grpc
+
 # TODO: Remove this context and filter once the protobuf UserWarning issue is downgraded to INFO
 # https://github.com/grpc/grpc/issues/37609
 with warnings.catch_warnings():
@@ -43,7 +45,12 @@
 from beartype import beartype as check_input_types
 import semver
 
-from ansys.api.dbu.v0.admin_pb2 import BackendType as GRPCBackendType
+from ansys.api.dbu.v0.admin_pb2 import (
+    BackendType as GRPCBackendType,
+    LogsRequest,
+    LogsTarget,
+    PeriodType,
+)
 from ansys.api.dbu.v0.admin_pb2_grpc import AdminStub
 from ansys.geometry.core.connection.backend import BackendType
 from ansys.geometry.core.connection.defaults import DEFAULT_HOST, DEFAULT_PORT, MAX_MESSAGE_LENGTH
@@ -327,3 +334,71 @@ def target(self) -> str:
     def get_name(self) -> str:
         """Get the target name of the connection."""
         return self._target
+
+    @check_input_types
+    @protect_grpc
+    def _get_service_logs(
+        self,
+        all_logs: bool = False,
+        dump_to_file: bool = False,
+        logs_folder: str | Path | None = None,
+    ) -> str | dict[str, str] | Path:
+        """Get the service logs.
+
+        Parameters
+        ----------
+        all_logs : bool, default: False
+            Flag indicating whether all logs should be retrieved. By default,
+            only the current logs are retrieved.
+        dump_to_file : bool, default: False
+            Flag indicating whether the logs should be dumped to a file.
+            By default, the logs are not dumped to a file.
+        logs_folder : str,  Path or None, default: None
+            Name of the folder where the logs should be dumped. This parameter
+            is only used if the ``dump_to_file`` parameter is set to ``True``.
+
+        Returns
+        -------
+        str
+            Service logs as a string. This is returned if the ``dump_to_file`` parameter
+            is set to ``False``.
+        dict[str, str]
+            Dictionary containing the logs. The keys are the logs names,
+            and the values are the logs as strings. This is returned if the ``all_logs``
+            parameter is set to ``True`` and the ``dump_to_file`` parameter
+            is set to ``False``.
+        Path
+            Path to the folder containing the logs (if the ``all_logs``
+            parameter is set to ``True``) or the path to the log file (if only
+            the current logs are retrieved). The ``dump_to_file`` parameter
+            must be set to ``True``.
+        """
+        request = LogsRequest(
+            target=LogsTarget.CLIENT,
+            period_type=PeriodType.CURRENT if not all_logs else PeriodType.ALL,
+            null_path=None,
+            null_period=None,
+        )
+        logs_generator = self._admin_stub.GetLogs(request)
+        logs: dict[str, str] = {}
+
+        for chunk in logs_generator:
+            if chunk.log_name not in logs:
+                logs[chunk.log_name] = ""
+            logs[chunk.log_name] += chunk.log_chunk.decode()
+
+        # Let's handle the various scenarios...
+        if not dump_to_file:
+            return logs if all_logs else next(iter(logs.values()))
+        else:
+            if logs_folder is None:
+                logs_folder = Path.cwd()
+            elif isinstance(logs_folder, str):
+                logs_folder = Path(logs_folder)
+
+            logs_folder.mkdir(parents=True, exist_ok=True)
+            for log_name, log_content in logs.items():
+                with (logs_folder / log_name).open("w") as f:
+                    f.write(log_content)
+
+            return (logs_folder / log_name) if len(logs) == 1 else logs_folder

src/ansys/geometry/core/modeler.py

@@ -497,3 +497,44 @@ def prepare_tools(self) -> PrepareTools:
     def measurement_tools(self) -> MeasurementTools:
         """Access to measurement tools."""
         return self._measurement_tools
+
+    @min_backend_version(25, 1, 0)
+    def get_service_logs(
+        self,
+        all_logs: bool = False,
+        dump_to_file: bool = False,
+        logs_folder: str | Path | None = None,
+    ) -> str | dict[str, str] | Path:
+        """Get the service logs.
+
+        Parameters
+        ----------
+        all_logs : bool, default: False
+            Flag indicating whether all logs should be retrieved. By default,
+            only the current logs are retrieved.
+        dump_to_file : bool, default: False
+            Flag indicating whether the logs should be dumped to a file.
+            By default, the logs are not dumped to a file.
+        logs_folder : str,  Path or None, default: None
+            Name of the folder where the logs should be dumped. This parameter
+            is only used if the ``dump_to_file`` parameter is set to ``True``.
+
+        Returns
+        -------
+        str
+            Service logs as a string. This is returned if the ``dump_to_file`` parameter
+            is set to ``False``.
+        dict[str, str]
+            Dictionary containing the logs. The keys are the logs names,
+            and the values are the logs as strings. This is returned if the ``all_logs``
+            parameter is set to ``True`` and the ``dump_to_file`` parameter
+            is set to ``False``.
+        Path
+            Path to the folder containing the logs (if the ``all_logs``
+            parameter is set to ``True``) or the path to the log file (if only
+            the current logs are retrieved). The ``dump_to_file`` parameter
+            must be set to ``True``.
+        """
+        return self.client._get_service_logs(
+            all_logs=all_logs, dump_to_file=dump_to_file, logs_folder=logs_folder
+        )

tests/integration/test_client.py

@@ -21,6 +21,8 @@
 # SOFTWARE.
 """Test basic client connection."""
 
+from pathlib import Path
+
 from grpc import insecure_channel
 import pytest
 
@@ -65,3 +67,32 @@ def test_client_close(client: GrpcClient):
     assert not client.healthy
     assert "Closed" in str(client)
     assert client.target() == ""
+
+
+def test_client_get_service_logs(client: GrpcClient):
+    """Test the retrieval of the service logs."""
+    # Low level call
+    logs = client._get_service_logs()
+    assert isinstance(logs, str)
+    assert logs  # is not empty
+
+    # Let's request them again on file dump
+    logs_folder = str(Path(__file__).parent / "logs")
+    logs_file_dump = client._get_service_logs(dump_to_file=True, logs_folder=logs_folder)
+    assert logs_file_dump.exists()
+
+    # Do not provide a folder
+    logs_file_dump = client._get_service_logs(dump_to_file=True)
+    assert logs_file_dump.exists()
+    logs_file_dump.unlink()  # Delete the file
+
+    # Let's request all logs now
+    logs_all = client._get_service_logs(all_logs=True)
+    assert isinstance(logs_all, dict)
+    assert logs_all  # is not empty
+
+    # Let's do the same directly from a Modeler object
+    modeler = Modeler(channel=client.channel)
+    logs_modeler = modeler.get_service_logs()
+    assert isinstance(logs_modeler, str)
+    assert logs_modeler  # is not empty