Coverage for gws-app/gws/lib/osx/__init__.py: 77%
208 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +0200
1"""Operating system and shell utilities.
3This package wraps common operating system tasks used throughout GWS:
5- running external commands (``run``, ``run_nowait``),
6- file system operations (``unlink``, ``rename``, ``copy``, ``rmdir``, ``touch``, ``chown``),
7- file information (``file_mtime``, ``file_age``, ``file_size``, ``file_checksum``),
8- searching directories (``find_files``, ``find_directories``),
9- path manipulation (``parse_path``, ``abs_path``, ``rel_path``, ``abs_web_path``),
10- processes and users (``kill_pid``, ``running_pids``, ``process_rss_size``, ``user_info``).
12Most file functions accept paths as ``str`` or ``bytes``. Functions that query
13files return a sentinel value (``-1`` or an empty string) instead of raising
14when the file cannot be accessed.
16Example::
18 import gws.lib.osx
20 out = gws.lib.osx.run(['gdalinfo', '--version'])
21 for path in gws.lib.osx.find_files('/data/projects', ext='json'):
22 print(path, gws.lib.osx.file_size(path))
23"""
25from typing import Optional
27import grp
28import hashlib
29import os
30import pwd
31import re
32import shlex
33import shutil
34import signal
35import subprocess
36import time
38import psutil
40import gws
43class Error(gws.Error):
44 """Generic error raised by OS utilities."""
46 pass
49class TimeoutError(Error):
50 """Raised when an external command times out."""
52 pass
55_Path = str | bytes
58def getenv(key: str, default: str = None) -> Optional[str]:
59 """Return the value of an environment variable.
61 Args:
62 key: Variable name.
63 default: Value to return if the variable is not set.
65 Returns:
66 The variable value, or ``default`` if the variable is not set.
67 """
68 return os.getenv(key, default)
71def run_nowait(cmd: str | list, **kwargs) -> subprocess.Popen:
72 """Start a process and return immediately, without waiting for it to finish.
74 By default, the process inherits stdin, stdout and stderr, and the command is not run in a shell.
76 Args:
77 cmd: Command to run, as a string or a list of arguments.
78 kwargs: Arguments to pass to ``subprocess.Popen``.
80 Returns:
81 The ``subprocess.Popen`` object of the started process.
82 """
84 args = {
85 'stdin': None,
86 'stdout': None,
87 'stderr': None,
88 'shell': False,
89 }
90 args.update(kwargs)
92 return subprocess.Popen(cmd, **args)
95def run(cmd: str | list, input: str = None, echo: bool = False, strict: bool = True, timeout: float = None, **kwargs) -> str:
96 """Run an external command and wait for it to finish.
98 A string command is split into arguments with ``shlex.split``. The command is not run in a shell,
99 and stderr is merged into stdout.
101 Args:
102 cmd: Command to run, as a string or a list of arguments.
103 input: Data to send to the command's stdin.
104 echo: Let the output go to the console instead of capturing it.
105 strict: Raise an error on a non-zero exit code.
106 timeout: Timeout in seconds.
107 kwargs: Arguments to pass to ``subprocess.Popen``.
109 Returns:
110 The captured command output, or an empty string if the output was not captured.
112 Raises:
113 ``TimeoutError``: If the command times out.
114 ``Error``: If the command cannot be run, or exits with a non-zero code and ``strict`` is true.
115 """
117 args = {
118 'stdin': subprocess.PIPE if input else None,
119 'stdout': None if echo else subprocess.PIPE,
120 'stderr': subprocess.STDOUT,
121 'shell': False,
122 }
123 args.update(kwargs)
125 if isinstance(cmd, str):
126 cmd = shlex.split(cmd)
128 gws.log.debug(f'RUN: {cmd=}')
130 try:
131 p = subprocess.Popen(cmd, **args)
132 out, _ = p.communicate(input, timeout)
133 rc = p.returncode
134 except subprocess.TimeoutExpired as exc:
135 raise TimeoutError(f'run: command timed out', repr(cmd)) from exc
136 except Exception as exc:
137 raise Error(f'run: failure', repr(cmd)) from exc
139 if rc:
140 gws.log.debug(f'RUN_FAILED: {cmd=} {rc=} {out=}')
142 if rc and strict:
143 raise Error(f'run: non-zero exit', repr(cmd))
145 return _to_str(out or '')
148def unlink(path: _Path) -> bool:
149 """Delete a file.
151 Directories and non-existing paths are ignored.
153 Args:
154 path: File path.
156 Returns:
157 ``True`` on success or if there was nothing to delete, ``False`` if an OS error occurred.
158 """
159 try:
160 if os.path.isfile(path):
161 os.unlink(path)
162 return True
163 except OSError as exc:
164 gws.log.debug(f'OSError: unlink: {exc}')
165 return False
168def rename(src: _Path, dst: _Path):
169 """Move or rename a file or directory.
171 Args:
172 src: Source path.
173 dst: Destination path.
174 """
176 shutil.move(_to_str(src), _to_str(dst))
179def chown(path: _Path, user: int = None, group: int = None):
180 """Change the owner and group of a path.
182 Args:
183 path: File path.
184 user: User ID, defaults to ``gws.c.UID``.
185 group: Group ID, defaults to ``gws.c.GID``.
186 """
187 os.chown(path, user or gws.c.UID, group or gws.c.GID)
190def copy(src: _Path, dst: _Path, user: int = None, group: int = None):
191 """Copy a file and set the owner of the copy.
193 Args:
194 src: Source path.
195 dst: Destination path.
196 user: User ID of the copy, defaults to ``gws.c.UID``.
197 group: Group ID of the copy, defaults to ``gws.c.GID``.
198 """
199 shutil.copyfile(src, dst)
200 os.chown(dst, user or gws.c.UID, group or gws.c.GID)
203def rmdir(path: _Path) -> bool:
204 """Remove a directory or a directory tree.
206 Args:
207 path: Path to a directory. Can be non-empty.
209 Returns:
210 ``True`` if the directory was removed, ``False`` if it does not exist or an OS error occurred.
211 """
213 if not os.path.isdir(path):
214 return False
215 try:
216 shutil.rmtree(path)
217 return True
218 except OSError as exc:
219 gws.log.warning(f'OSError: rmdir: {exc}')
220 return False
223def touch(path: _Path):
224 """Set the access and modification times of a file to the current time.
226 If the file does not exist, it is created.
228 Args:
229 path: File path.
230 """
231 with open(path, 'a'):
232 os.utime(path, None)
235def file_mtime(path: _Path) -> float:
236 """Return the modification time of a path.
238 Args:
239 path: File or directory path.
241 Returns:
242 Modification time in seconds since the epoch, or ``-1`` if the path cannot be accessed.
243 """
244 try:
245 return os.stat(path).st_mtime
246 except OSError:
247 return -1
250def file_age(path: _Path) -> int:
251 """Return the number of seconds since a path was last modified.
253 Args:
254 path: File path.
256 Returns:
257 Age in seconds, or ``-1`` if the path cannot be accessed.
258 """
259 try:
260 return int(time.time() - os.stat(path).st_mtime)
261 except OSError:
262 return -1
265def file_size(path: _Path) -> int:
266 """Return the size of a file.
268 Args:
269 path: File path.
271 Returns:
272 Size in bytes, or ``-1`` if the path cannot be accessed.
273 """
274 try:
275 return os.stat(path).st_size
276 except OSError:
277 return -1
280def file_checksum(path: _Path) -> str:
281 """Return the SHA-256 checksum of a file.
283 Args:
284 path: File path.
286 Returns:
287 Hex digest of the file content, or an empty string if the file cannot be read.
288 """
289 try:
290 with open(path, 'rb') as fp:
291 return hashlib.sha256(fp.read()).hexdigest()
292 except OSError as exc:
293 return ''
296def kill_pid(pid: int, sig_name='TERM') -> bool:
297 """Send a signal to a process.
299 Args:
300 pid: Process ID.
301 sig_name: Signal name, with or without the ``SIG`` prefix, e.g. ``TERM`` or ``SIGKILL``.
303 Returns:
304 ``True`` if the signal was sent or the process does not exist, ``False`` if the signal could not be sent.
305 """
306 sig = getattr(signal, sig_name, None) or getattr(signal, 'SIG' + sig_name)
307 try:
308 psutil.Process(pid).send_signal(sig)
309 return True
310 except psutil.NoSuchProcess:
311 return True
312 except psutil.Error as e:
313 gws.log.warning(f'send_signal failed, pid={pid!r}, {e}')
314 return False
317def running_pids() -> dict[int, str]:
318 """Return all running processes.
320 Returns:
321 A dict mapping process IDs to process names.
322 """
323 d = {}
324 for p in psutil.process_iter():
325 d[p.pid] = p.name()
326 return d
329def process_rss_size(unit: str = 'm') -> float:
330 """Return the Resident Set Size of the current process.
332 Args:
333 unit: ``k`` (kilobytes), ``m`` (megabytes) or ``g`` (gigabytes). Any other value returns bytes.
335 Returns:
336 The Resident Set Size in the given unit.
337 """
338 n = psutil.Process().memory_info().rss
339 if unit == 'k':
340 return n / 1e3
341 if unit == 'm':
342 return n / 1e6
343 if unit == 'g':
344 return n / 1e9
345 return n
348def user_info(uid=None, gid=None) -> dict:
349 """Return user and group information.
351 Args:
352 uid: User ID. Defaults to the user ID of the current process.
353 gid: Group ID. Defaults to the user's primary group ID.
355 Returns:
356 A dict with the keys ``pw_name``, ``pw_uid``, ``pw_gid``, ``pw_dir``, ``pw_shell``
357 (from ``struct_passwd``) and ``gr_name``, ``gr_gid`` (from ``struct_group``).
358 """
360 uid = uid or os.getuid()
361 u = pwd.getpwuid(uid)
363 r = dict(
364 pw_name=u.pw_name,
365 pw_uid=u.pw_uid,
366 pw_gid=u.pw_gid,
367 pw_dir=u.pw_dir,
368 pw_shell=u.pw_shell,
369 )
371 gid = gid or r['pw_gid']
372 g = grp.getgrgid(gid)
374 r['gr_name'] = g.gr_name
375 r['gr_gid'] = g.gr_gid
377 return r
380def find_entries(dirname: _Path, deep: bool = True):
381 """Find entries in a directory, skipping hidden ones.
383 Args:
384 dirname: Path to a directory.
385 deep: If true, also search subdirectories recursively.
387 Yields:
388 ``os.DirEntry`` objects for files and directories whose names do not start with a dot.
389 """
390 de: os.DirEntry
391 for de in os.scandir(dirname):
392 if de.name.startswith('.'):
393 continue
394 yield de
395 if de.is_dir() and deep:
396 yield from find_entries(de.path, deep=deep)
399def find_files(dirname: _Path, pattern=None, ext=None, deep: bool = True):
400 """Find files in a directory, skipping hidden ones.
402 Args:
403 dirname: Path to a directory.
404 pattern: Regular expression to search for in the file path.
405 ext: Extension or a list of extensions to match. Used only if ``pattern`` is not given.
406 deep: If true, also search subdirectories recursively.
408 Yields:
409 Paths of matching files.
410 """
411 if not pattern and ext:
412 if isinstance(ext, (list, tuple)):
413 ext = '|'.join(ext)
414 pattern = '\\.(' + ext + ')$'
416 for de in find_entries(dirname, deep=deep):
417 if de.is_file() and (pattern is None or re.search(pattern, de.path)):
418 yield de.path
421def find_directories(dirname: _Path, pattern=None, deep: bool = True):
422 """Find directories in a directory, skipping hidden ones.
424 Args:
425 dirname: Path to a directory.
426 pattern: Regular expression to search for in the directory path.
427 deep: If true, also search subdirectories recursively.
429 Yields:
430 Paths of matching directories.
431 """
433 for de in find_entries(dirname, deep=deep):
434 if de.is_dir() and (pattern is None or re.search(pattern, de.path)):
435 yield de.path
438class ParsePathResult(gws.Data):
439 """Components of a file path, as returned by ``parse_path``."""
441 path: str
442 """The full path."""
443 dirname: str
444 """The directory part."""
445 filename: str
446 """The file name, including the extension."""
447 stem: str
448 """The file name up to the first dot."""
449 extension: str
450 """The file name after the first dot, without the dot."""
453def parse_path(path: _Path) -> ParsePathResult:
454 """Split a file path into its components.
456 The extension is everything after the first dot in the file name, so ``a.tar.gz``
457 has the stem ``a`` and the extension ``tar.gz``. A file name starting with a dot has no extension.
459 Args:
460 path: File path.
462 Returns:
463 The path components.
464 """
466 str_path = _to_str(path)
467 sp = os.path.split(str_path)
469 pp = ParsePathResult(
470 path=str_path,
471 dirname='',
472 filename='',
473 stem='',
474 extension='',
475 )
476 pp.dirname = sp[0]
477 pp.filename = sp[1]
479 if pp.filename.startswith('.'):
480 pp.stem = pp.filename
481 else:
482 pp.stem, _, pp.extension = pp.filename.partition('.')
484 return pp
487def file_name(path: _Path) -> str:
488 """Return the file name part of a path.
490 Args:
491 path: File path.
493 Returns:
494 The last component of the path.
495 """
497 sp = os.path.split(_to_str(path))
498 return sp[1]
501def is_abs_path(path: _Path) -> bool:
502 """Check if a path is absolute.
504 Args:
505 path: File path.
507 Returns:
508 ``True`` if the path is absolute.
509 """
510 return os.path.isabs(path)
513def abs_path(path: _Path, base: _Path) -> str:
514 """Make a relative path absolute with respect to a base directory or file path.
516 If ``base`` is a file, its directory is used. An absolute ``path`` is only normalized.
518 Args:
519 path: A path.
520 base: Base directory or file path.
522 Returns:
523 The absolute path.
525 Raises:
526 ``ValueError``: If ``path`` is relative and ``base`` is empty.
527 """
529 str_path = _to_str(path)
531 if os.path.isabs(str_path):
532 return os.path.normpath(str_path)
534 if not base:
535 raise ValueError('cannot compute abspath without a base')
537 if os.path.isfile(base):
538 base = os.path.dirname(base)
540 return os.path.abspath(os.path.join(_to_str(base), str_path))
543def abs_web_path(path: str, basedir: str) -> Optional[str]:
544 """Resolve a web path in a base directory.
546 The path components must consist of letters, digits, ``_`` and ``-``,
547 the file name can also contain lowercase extensions. This prevents path traversal.
549 Args:
550 path: Slash-separated path, as received from a web request.
551 basedir: Path to the base directory.
553 Returns:
554 The path of an existing file in the base directory, or ``None`` if the path is invalid or the file does not exist.
555 """
557 _dir_re = r'^[A-Za-z0-9_-]+$'
558 _fil_re = r'^[A-Za-z0-9_-]+(\.[a-z0-9]+)*$'
560 gws.log.debug(f'abs_web_path: trying {path!r} in {basedir!r}')
562 dirs = []
563 for s in path.split('/'):
564 s = s.strip()
565 if s:
566 dirs.append(s)
568 if not dirs:
569 gws.log.warning(f'abs_web_path: empty path={path!r}')
570 return
572 fname = dirs.pop()
574 if not all(re.match(_dir_re, p) for p in dirs):
575 gws.log.warning(f'abs_web_path: invalid dirname in path={path!r}')
576 return
578 if not re.match(_fil_re, fname):
579 gws.log.warning(f'abs_web_path: invalid filename in path={path!r}')
580 return
582 p = basedir
583 if dirs:
584 p += '/' + '/'.join(dirs)
585 p += '/' + fname
587 if not os.path.isfile(p):
588 gws.log.warning(f'abs_web_path: not a file path={path!r}')
589 return
591 return p
594def rel_path(path: _Path, base: _Path) -> str:
595 """Make a path relative to a base directory or file path.
597 If ``base`` is a file, its directory is used.
599 Args:
600 path: Path to make relative.
601 base: Base directory or file path.
603 Returns:
604 The relative path.
605 """
607 if os.path.isfile(base):
608 base = os.path.dirname(base)
610 return os.path.relpath(_to_str(path), _to_str(base))
613def _to_str(p: _Path) -> str:
614 return p if isinstance(p, str) else bytes(p).decode('utf8')
617def _to_bytes(p: _Path) -> bytes:
618 return p if isinstance(p, bytes) else str(p).encode('utf8')