Add functionality to get specific stdout or stderr position and length (2c4821ba) · Commits · NDIP / Nova Packages / nova-galaxy

src/nova/galaxy/job.py

+3 −4

Original line number	Diff line number	Diff line
		"""Internal job related classes and functions."""

		import sys
		import time
		from threading import Thread
		from typing import TYPE_CHECKING, Dict, Optional
		@@ -193,12 +192,12 @@ class Job:
		time.sleep(1)
		return None

		def get_console_output(self) -> Dict[str, str]:
		def get_console_output(self, start: int, length: int) -> Dict[str, str]:
		"""Get all the current console output."""
		out = self.galaxy_instance.make_get_request(
		f"{self.store.nova_connection.galaxy_url}/api/jobs/"
		f"{self.id}/console_output?stdout_position=0&stdout_length="
		f"{sys.maxsize - 1}&stderr_position=0&stderr_length={sys.maxsize - 1}"
		f"{self.id}/console_output?stdout_position={start}&stdout_length="
		f"{length}&stderr_position={start}&stderr_length={length}"
		)
		out.raise_for_status()
		return out.json()

src/nova/galaxy/tool.py

+23 −4

Original line number	Diff line number	Diff line
		"""Contains classes to run tools in Galaxy via Connection."""

		import sys
		from typing import TYPE_CHECKING, List, Optional, Union

		if TYPE_CHECKING:
		@@ -145,32 +146,50 @@ class Tool(AbstractWork):
		if self._job:
		self._job.cancel(check_results=False)

		def get_stdout(self) -> Optional[str]:
		def get_stdout(self, position: Optional[int] = None, length: Optional[int] = None) -> Optional[str]:
		"""Get the current STDOUT for a tool.

		Will be overridden everytime this tool is run.

		Parameters
		----------
		position: int, optional
		The starting position from which to get the stdout. Useful if not wanting to fetch the entire stdout file.
		length: int, optional
		How many characters of stdout to read.

		Returns
		-------
		Optional[str]
		Returns the current STDOUT of the tool if it is running or finished.
		"""
		if self._job:
		return self._job.get_console_output()["stdout"]
		_position = 0 if position is None else position
		_length = sys.maxsize - 1 if length is None else length
		return self._job.get_console_output(_position, _length)["stdout"]
		return None

		def get_stderr(self) -> Optional[str]:
		def get_stderr(self, position: Optional[int] = None, length: Optional[int] = None) -> Optional[str]:
		"""Get the current STDERR for a tool.

		Will be overridden everytime this tool is run.

		Parameters
		----------
		position: int, optional
		The starting position from which to get the stderr. Useful if not wanting to fetch the entire stderr file.
		length: int, optional
		How many characters of stdout to read.

		Returns
		-------
		Optional[str]
		Returns the current STDERR of the tool if it is running or finished.
		"""
		if self._job:
		return self._job.get_console_output()["stderr"]
		_position = 0 if position is None else position
		_length = sys.maxsize - 1 if length is None else length
		return self._job.get_console_output(_position, _length)["stderr"]
		return None

		def get_url(self, max_tries: int = 5, check_url: bool = False) -> Optional[str]:

tests/test_run_tool.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -100,7 +100,9 @@ def test_get_tool_stdout(nova_instance: Connection) -> None:
		assert state == WorkState.RUNNING
		time.sleep(10) # Tool takes a moment to produce stdout
		stdout = test_tool.get_stdout()
		stdout_substring = test_tool.get_stdout(5, 50)
		assert stdout is not None
		assert stdout_substring in stdout
		test_tool.cancel()