SDK-2974: add process info tools

src/galaxy/api/types.py

@@ -61,7 +61,6 @@ class NextStep():
     :param auth_params: configuration options: {"window_title": :class:`str`, "window_width": :class:`str`, "window_height": :class:`int`, "start_uri": :class:`int`, "end_uri_regex": :class:`str`}
     :param cookies: browser initial set of cookies
     :param js: a map of the url regex patterns into the list of *js* scripts that should be executed on every document at given step of internal browser authentication.
-
     """
     next_step: str
     auth_params: Dict[str, str]

src/galaxy/proc_tools.py

@@ -0,0 +1,91 @@
+import platform
+from dataclasses import dataclass
+from typing import Iterable, NewType, Optional, Set
+
+
+def is_windows():
+    return platform.system() == "Windows"
+
+
+ProcessId = NewType("ProcessId", int)
+
+
+@dataclass
+class ProcessInfo:
+    pid: ProcessId
+    binary_path: Optional[str]
+
+
+if is_windows():
+    from ctypes import byref, sizeof, windll, create_unicode_buffer, FormatError, WinError
+    from ctypes.wintypes import DWORD
+
+
+    def pids() -> Iterable[ProcessId]:
+        _PROC_ID_T = DWORD
+        list_size = 4096
+
+        def try_get_pids(list_size: int) -> Set[ProcessId]:
+            result_size = DWORD()
+            proc_id_list = (_PROC_ID_T * list_size)()
+
+            if not windll.psapi.EnumProcesses(byref(proc_id_list), sizeof(proc_id_list), byref(result_size)):
+                raise WinError(descr="Failed to get process ID list: %s" % FormatError())
+
+            return proc_id_list[:int(result_size.value / sizeof(_PROC_ID_T()))]
+
+        while True:
+            proc_ids = try_get_pids(list_size)
+            if len(proc_ids) < list_size:
+                return proc_ids
+
+            list_size *= 2
+
+
+    def get_process_info(pid: ProcessId) -> Optional[ProcessInfo]:
+        _PROC_QUERY_LIMITED_INFORMATION = 0x1000
+
+        process_info = ProcessInfo(pid=pid, binary_path=None)
+
+        h_process = windll.kernel32.OpenProcess(_PROC_QUERY_LIMITED_INFORMATION, False, pid)
+        if not h_process:
+            return process_info
+
+        try:
+            def get_exe_path() -> Optional[str]:
+                _MAX_PATH = 260
+                _WIN32_PATH_FORMAT = 0x0000
+
+                exe_path_buffer = create_unicode_buffer(_MAX_PATH)
+                exe_path_len = DWORD(len(exe_path_buffer))
+
+                return exe_path_buffer[:exe_path_len.value] if windll.kernel32.QueryFullProcessImageNameW(
+                    h_process, _WIN32_PATH_FORMAT, exe_path_buffer, byref(exe_path_len)
+                ) else None
+
+            process_info.binary_path = get_exe_path()
+        finally:
+            windll.kernel32.CloseHandle(h_process)
+            return process_info
+else:
+    import psutil
+
+
+    def pids() -> Iterable[ProcessId]:
+        for pid in psutil.pids():
+            yield pid
+
+
+    def get_process_info(pid: ProcessId) -> Optional[ProcessInfo]:
+        process_info = ProcessInfo(pid=pid, binary_path=None)
+        try:
+            process_info.binary_path = psutil.Process(pid=pid).as_dict(attrs=["exe"])["exe"]
+        except psutil.NoSuchProcess:
+            pass
+        finally:
+            return process_info
+
+
+def process_iter() -> Iterable[ProcessInfo]:
+    for pid in pids():
+        yield get_process_info(pid)

src/galaxy/tools.py

@@ -3,6 +3,7 @@ import os
 import zipfile
 from glob import glob
 
+
 def zip_folder(folder):
     files = glob(os.path.join(folder, "**"), recursive=True)
     files = [file.replace(folder + os.sep, "") for file in files]
@@ -14,6 +15,7 @@ def zip_folder(folder):
             zipf.write(os.path.join(folder, file), arcname=file)
     return zip_buffer
 
+
 def zip_folder_to_file(folder, filename):
     zip_content = zip_folder(folder).getbuffer()
     with open(filename, "wb") as archive: