feat: api improved following code review

Author: ymarcon

datashield/api.py

@@ -2,6 +2,7 @@
 DataSHIELD API.
 """
 
+import logging
 from datashield.interface import DSLoginInfo, DSConnection, DSDriver, DSError
 import time
 
@@ -72,13 +73,17 @@ class DSSession:
     DataSHIELD session, establishes connections with remote servers and performs commands.
     """
 
-    def __init__(self, logins: list[DSLoginInfo]):
+    def __init__(self, logins: list[DSLoginInfo], start_timeout: float = 300.0, start_delay: float = 0.1):
         """
         Create a session, with connection information. Does not open the connections.
 
         :param logins: A list of login details
+        :param start_timeout: The maximum time in seconds to wait for R sessions to start, default is 300 seconds (5 minutes)
+        :param start_delay: The delay in seconds between checking if R sessions are started, default is 0.1 seconds
         """
         self.logins = logins
+        self.start_timeout = start_timeout
+        self.start_delay = start_delay
         self.conns: list[DSConnection] = None
         self.errors: dict = None
 
@@ -234,61 +239,108 @@ def workspaces(self) -> dict:
             rval[conn.name] = conn.list_workspaces()
         return rval
 
-    def workspace_save(self, name: str) -> dict:
+    def workspace_save(self, name: str) -> None:
         """
         Save the DataSHIELD R session in a workspace on the remote data repository.
 
         :param name: The name of the workspace
-        :return: The list of DataSHIELD workspaces, that have been saved on the remote data repository after saving the workspace, per remote server name
         """
         for conn in self.conns:
             conn.save_workspace(f"{conn.name}:{name}")
-        return self.workspaces()
 
-    def workspace_restore(self, name: str) -> dict:
+    def workspace_restore(self, name: str) -> None:
         """
         Restore a saved DataSHIELD R session from the remote data repository. When restoring a workspace,
         any existing symbol or file with same name will be overridden.
 
         :param name: The name of the workspace
-        :return: The list of DataSHIELD workspaces, that have been saved on the remote data repository after restoring the workspace, per remote server name
         """
         for conn in self.conns:
             conn.restore_workspace(f"{conn.name}:{name}")
-        return self.workspaces()
 
-    def workspace_rm(self, name: str) -> dict:
+    def workspace_rm(self, name: str) -> None:
         """
         Remove a DataSHIELD workspace from the remote data repository. Ignored if no
         such workspace exists.
 
         :param name: The name of the workspace
-        :return: The list of DataSHIELD workspaces, that have been saved on the remote data repository after removing the workspace, per remote server name
         """
         for conn in self.conns:
             conn.rm_workspace(f"{conn.name}:{name}")
-        return self.workspaces()
 
     #
     # R session
     #
 
     def sessions(self) -> dict:
         """
-        Ensure R sessions are started on the remote servers and get their information.
-
-        :return: The R session information, per remote server name
+        Ensure R sessions are started on the remote servers and wait until they are ready.
+        This method returns a dictionary mapping each remote server name to its underlying
+        R session object (an ``RSession`` instance). These session objects are
+        primarily intended for status inspection (e.g. ``is_started()``, ``is_ready()``,
+        ``is_pending()``, ``is_failed()``, ``is_terminated()``, ``get_last_message()``) and
+        not for direct interaction with the remote R environment.
+        In normal use, you do not need to work with the returned session objects directly.
+        Instead, interact with the remote R sessions through the higher-level ``DSSession``
+        methods (such as assignment, aggregation, workspace and other helpers), which
+        operate on all underlying sessions. Important: only sessions that have been successfully started
+        and are ready will be included in the returned dictionary and used for subsequent operations.
+        If a session fails to start or check status, it will be excluded from the returned dictionary
+        and from subsequent operations, and an error will be logged. If no sessions can be started successfully,
+        an exception will be raised.
+
+        :return: A dictionary mapping remote server names to their corresponding R session
+            objects, intended mainly for internal use and status monitoring.
         """
         rval = {}
+        self._init_errors()
+        started_conns = []
+        excluded_conns = []
+
+        # start sessions asynchronously if supported, otherwise synchronously
         for conn in self.conns:
-            if not conn.has_session():
-                conn.start_session(asynchronous=True)
-        # check for session status and wait until all are complete
-        while any(conn.get_session().is_pending() for conn in self.conns):
-            time.sleep(0.1)
+            try:
+                if not conn.has_session():
+                    conn.start_session(asynchronous=True)
+            except Exception as e:
+                logging.warning(f"Failed to start session: {conn.name} - {e}")
+                excluded_conns.append(conn.name)
+
+        # check for session status and wait until all are started
+        for conn in [c for c in self.conns if c.name not in excluded_conns]:
+            try:
+                if conn.is_session_started():
+                    started_conns.append(conn.name)
+            except Exception as e:
+                logging.warning(f"Failed to check session status: {conn.name} - {e}")
+                excluded_conns.append(conn.name)
+
+        # wait until all sessions are started, excluding those that have failed to start or check status
+        start_time = time.time()
+        while len(started_conns) < len(self.conns) - len(excluded_conns):
+            if time.time() - start_time > self.start_timeout:
+                raise DSError("Timed out waiting for R sessions to start")
+            time.sleep(self.start_delay)
+            remaining_conns = [
+                conn for conn in self.conns if conn.name not in started_conns and conn.name not in excluded_conns
+            ]
+            for conn in remaining_conns:
+                try:
+                    if conn.is_session_started():
+                        started_conns.append(conn.name)
+                except Exception as e:
+                    logging.warning(f"Failed to check session status: {conn.name} - {e}")
+                    excluded_conns.append(conn.name)
+
+        # at this point, all sessions that could be started have been started, and those that failed to start or check status have been excluded
         for conn in self.conns:
-            rval[conn.name] = conn.get_session()
-        self._check_errors()
+            if conn.name in started_conns:
+                rval[conn.name] = conn.get_session()
+        if len(excluded_conns) > 0:
+            logging.error(f"Some sessions have been excluded due to errors: {', '.join(excluded_conns)}")
+            self.conns = [conn for conn in self.conns if conn.name not in excluded_conns]
+        if len(self.conns) == 0:
+            raise DSError("No sessions could be started successfully.")
         return rval
 
     def ls(self) -> dict:
@@ -297,8 +349,8 @@ def ls(self) -> dict:
 
         :return: The symbols that live in the DataSHIELD R session on the server side, per remote server name
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         rval = {}
         for conn in self.conns:
             try:
@@ -315,8 +367,8 @@ def rm(self, symbol: str) -> None:
 
         :param symbol: The name of the symbol to remove
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         for conn in self.conns:
             try:
                 conn.rm_symbol(symbol)
@@ -343,8 +395,8 @@ def assign_table(
         :param tables: The name of the table to assign, per server name. If not defined, 'table' is used.
         :param asynchronous: Whether the operation is asynchronous (if supported by the DataSHIELD server)
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         cmd = {}
         for conn in self.conns:
             name = table
@@ -370,8 +422,8 @@ def assign_resource(
         :param resources: The name of the resource to assign, per server name. If not defined, 'resource' is used.
         :param asynchronous: Whether the operation is asynchronous (if supported by the DataSHIELD server)
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         cmd = {}
         for conn in self.conns:
             name = resource
@@ -394,8 +446,8 @@ def assign_expr(self, symbol: str, expr: str, asynchronous: bool = True) -> None
         :param expr: The R expression to evaluate and which result will be assigned
         :param asynchronous: Whether the operation is asynchronous (if supported by the DataSHIELD server)
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         cmd = {}
         for conn in self.conns:
             try:
@@ -415,8 +467,8 @@ def aggregate(self, expr: str, asynchronous: bool = True) -> dict:
         :param asynchronous: Whether the operation is asynchronous (if supported by the DataSHIELD server)
         :return: The result of the aggregation expression evaluation, per remote server name
         """
-        self._init_errors()
-        self.sessions()  # ensure sessions are started and available
+        # ensure sessions are started and available
+        self.sessions()
         cmd = {}
         rval = {}
         for conn in self.conns:
@@ -465,6 +517,7 @@ def _append_error(self, conn: DSConnection, error: Exception) -> None:
         """
         Append an error.
         """
+        logging.error(f"[{conn.name}] {error}")
         self.errors[conn.name] = error
 
     def _check_errors(self) -> None:

datashield/interface.py

@@ -187,6 +187,18 @@ def start_session(self, asynchronous: bool = True) -> RSession:
         """
         raise NotImplementedError("DSConnection function not available")
 
+    def is_session_started(self) -> bool:
+        """
+        Get whether the session with the DataSHIELD server is started. If the session start was asynchronous, this function
+        can be used to check whether the session is started without waiting for it to be started. If the last call was positive,
+        subsequent calls will not request the server for session status, but will return True directly. If the last call was negative,
+        subsequent calls will request the server for session status until a positive response is obtained.
+
+        :return: Whether the session is started
+        :throws: DSError if the session was not started or session information is not available
+        """
+        raise NotImplementedError("DSConnection function not available")
+
     def get_session(self) -> RSession:
         """
         Get the R session with the DataSHIELD server. If no session is established, an error will be raised.
@@ -359,9 +371,10 @@ def is_async(self) -> dict:
         the raw result can be accessed asynchronously, allowing parallelization of DataSHIELD calls
         over multpile servers. The returned named list of logicals will specify if asynchronicity is supported for:
         aggregation operation ('aggregate'), table assignment operation ('assign_table'),
-        resource assignment operation ('assign_resource') and expression assignment operation ('assign_expr').
+        resource assignment operation ('assign_resource'), expression assignment operation ('assign_expr')
+        and R session creation ('session').
 
-        :return: A named list of logicals specifying if asynchronicity is supported for aggregation operation ('aggregate'), table assignment operation ('assign_table'),
+        :return: A named list of logicals specifying if asynchronicity is supported.
         """
         raise NotImplementedError("DSConnection function not available")