Coverage for gws-app/gws/lib/osx/__init__.py: 75%
210 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +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``, ``mkdir``, ``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 gws.lib.osx.mkdir('/tmp/gws/data')
22 for path in gws.lib.osx.find_files('/data/projects', ext='json'):
23 print(path, gws.lib.osx.file_size(path))
24"""
26from typing import Optional
28import grp
29import hashlib
30import os
31import pwd
32import re
33import shlex
34import shutil
35import signal
36import subprocess
37import time
39import psutil
41import gws
44class Error(gws.Error):
45 """Generic error raised by OS utilities."""
47 pass
50class TimeoutError(Error):
51 """Raised when an external command times out."""
53 pass
56_Path = str | bytes
59def getenv(key: str, default: str = None) -> Optional[str]:
60 """Return the value of an environment variable.
62 Args:
63 key: Variable name.
64 default: Value to return if the variable is not set.
66 Returns:
67 The variable value, or ``default`` if the variable is not set.
68 """
69 return os.getenv(key, default)
72def run_nowait(cmd: str | list, **kwargs) -> subprocess.Popen:
73 """Start a process and return immediately, without waiting for it to finish.
75 By default, the process inherits stdin, stdout and stderr, and the command is not run in a shell.
77 Args:
78 cmd: Command to run, as a string or a list of arguments.
79 kwargs: Arguments to pass to ``subprocess.Popen``.
81 Returns:
82 The ``subprocess.Popen`` object of the started process.
83 """
85 args = {
86 'stdin': None,
87 'stdout': None,
88 'stderr': None,
89 'shell': False,
90 }
91 args.update(kwargs)
93 return subprocess.Popen(cmd, **args)
96def run(cmd: str | list, input: str = None, echo: bool = False, strict: bool = True, timeout: float = None, **kwargs) -> str:
97 """Run an external command and wait for it to finish.
99 A string command is split into arguments with ``shlex.split``. The command is not run in a shell,
100 and stderr is merged into stdout.
102 Args:
103 cmd: Command to run, as a string or a list of arguments.
104 input: Data to send to the command's stdin.
105 echo: Let the output go to the console instead of capturing it.
106 strict: Raise an error on a non-zero exit code.
107 timeout: Timeout in seconds.
108 kwargs: Arguments to pass to ``subprocess.Popen``.
110 Returns:
111 The captured command output, or an empty string if the output was not captured.
113 Raises:
114 ``TimeoutError``: If the command times out.
115 ``Error``: If the command cannot be run, or exits with a non-zero code and ``strict`` is true.
116 """
118 args = {
119 'stdin': subprocess.PIPE if input else None,
120 'stdout': None if echo else subprocess.PIPE,
121 'stderr': subprocess.STDOUT,
122 'shell': False,
123 }
124 args.update(kwargs)
126 if isinstance(cmd, str):
127 cmd = shlex.split(cmd)
129 gws.log.debug(f'RUN: {cmd=}')
131 try:
132 p = subprocess.Popen(cmd, **args)
133 out, _ = p.communicate(input, timeout)
134 rc = p.returncode
135 except subprocess.TimeoutExpired as exc:
136 raise TimeoutError(f'run: command timed out', repr(cmd)) from exc
137 except Exception as exc:
138 raise Error(f'run: failure', repr(cmd)) from exc
140 if rc:
141 gws.log.debug(f'RUN_FAILED: {cmd=} {rc=} {out=}')
143 if rc and strict:
144 raise Error(f'run: non-zero exit', repr(cmd))
146 return _to_str(out or '')
149def unlink(path: _Path) -> bool:
150 """Delete a file.
152 Directories and non-existing paths are ignored.
154 Args:
155 path: File path.
157 Returns:
158 ``True`` on success or if there was nothing to delete, ``False`` if an OS error occurred.
159 """
160 try:
161 if os.path.isfile(path):
162 os.unlink(path)
163 return True
164 except OSError as exc:
165 gws.log.debug(f'OSError: unlink: {exc}')
166 return False
169def rename(src: _Path, dst: _Path):
170 """Move or rename a file or directory.
172 Args:
173 src: Source path.
174 dst: Destination path.
175 """
177 shutil.move(_to_str(src), _to_str(dst))
180def chown(path: _Path, user: int = None, group: int = None):
181 """Change the owner and group of a path.
183 Args:
184 path: File path.
185 user: User ID, defaults to ``gws.c.UID``.
186 group: Group ID, defaults to ``gws.c.GID``.
187 """
188 os.chown(path, user or gws.c.UID, group or gws.c.GID)
191def copy(src: _Path, dst: _Path, user: int = None, group: int = None):
192 """Copy a file and set the owner of the copy.
194 Args:
195 src: Source path.
196 dst: Destination path.
197 user: User ID of the copy, defaults to ``gws.c.UID``.
198 group: Group ID of the copy, defaults to ``gws.c.GID``.
199 """
200 shutil.copyfile(src, dst)
201 os.chown(dst, user or gws.c.UID, group or gws.c.GID)
204def mkdir(path: _Path, mode: int = 0o755, user: int = None, group: int = None):
205 """Create a directory, including missing parent directories.
207 Does nothing if the directory already exists.
209 Args:
210 path: Path to a directory.
211 mode: Directory creation mode.
212 user: Directory user. Currently not used.
213 group: Directory group. Currently not used.
214 """
216 os.makedirs(path, mode, exist_ok=True)
219def rmdir(path: _Path) -> bool:
220 """Remove a directory or a directory tree.
222 Args:
223 path: Path to a directory. Can be non-empty.
225 Returns:
226 ``True`` if the directory was removed, ``False`` if it does not exist or an OS error occurred.
227 """
229 if not os.path.isdir(path):
230 return False
231 try:
232 shutil.rmtree(path)
233 return True
234 except OSError as exc:
235 gws.log.warning(f'OSError: rmdir: {exc}')
236 return False
239def touch(path: _Path):
240 """Set the access and modification times of a file to the current time.
242 If the file does not exist, it is created.
244 Args:
245 path: File path.
246 """
247 with open(path, 'a'):
248 os.utime(path, None)
251def file_mtime(path: _Path) -> float:
252 """Return the modification time of a path.
254 Args:
255 path: File or directory path.
257 Returns:
258 Modification time in seconds since the epoch, or ``-1`` if the path cannot be accessed.
259 """
260 try:
261 return os.stat(path).st_mtime
262 except OSError:
263 return -1
266def file_age(path: _Path) -> int:
267 """Return the number of seconds since a path was last modified.
269 Args:
270 path: File path.
272 Returns:
273 Age in seconds, or ``-1`` if the path cannot be accessed.
274 """
275 try:
276 return int(time.time() - os.stat(path).st_mtime)
277 except OSError:
278 return -1
281def file_size(path: _Path) -> int:
282 """Return the size of a file.
284 Args:
285 path: File path.
287 Returns:
288 Size in bytes, or ``-1`` if the path cannot be accessed.
289 """
290 try:
291 return os.stat(path).st_size
292 except OSError:
293 return -1
296def file_checksum(path: _Path) -> str:
297 """Return the SHA-256 checksum of a file.
299 Args:
300 path: File path.
302 Returns:
303 Hex digest of the file content, or an empty string if the file cannot be read.
304 """
305 try:
306 with open(path, 'rb') as fp:
307 return hashlib.sha256(fp.read()).hexdigest()
308 except OSError as exc:
309 return ''
312def kill_pid(pid: int, sig_name='TERM') -> bool:
313 """Send a signal to a process.
315 Args:
316 pid: Process ID.
317 sig_name: Signal name, with or without the ``SIG`` prefix, e.g. ``TERM`` or ``SIGKILL``.
319 Returns:
320 ``True`` if the signal was sent or the process does not exist, ``False`` if the signal could not be sent.
321 """
322 sig = getattr(signal, sig_name, None) or getattr(signal, 'SIG' + sig_name)
323 try:
324 psutil.Process(pid).send_signal(sig)
325 return True
326 except psutil.NoSuchProcess:
327 return True
328 except psutil.Error as e:
329 gws.log.warning(f'send_signal failed, pid={pid!r}, {e}')
330 return False
333def running_pids() -> dict[int, str]:
334 """Return all running processes.
336 Returns:
337 A dict mapping process IDs to process names.
338 """
339 d = {}
340 for p in psutil.process_iter():
341 d[p.pid] = p.name()
342 return d
345def process_rss_size(unit: str = 'm') -> float:
346 """Return the Resident Set Size of the current process.
348 Args:
349 unit: ``k`` (kilobytes), ``m`` (megabytes) or ``g`` (gigabytes). Any other value returns bytes.
351 Returns:
352 The Resident Set Size in the given unit.
353 """
354 n = psutil.Process().memory_info().rss
355 if unit == 'k':
356 return n / 1e3
357 if unit == 'm':
358 return n / 1e6
359 if unit == 'g':
360 return n / 1e9
361 return n
364def user_info(uid=None, gid=None) -> dict:
365 """Return user and group information.
367 Args:
368 uid: User ID. Defaults to the user ID of the current process.
369 gid: Group ID. Defaults to the user's primary group ID.
371 Returns:
372 A dict with the keys ``pw_name``, ``pw_uid``, ``pw_gid``, ``pw_dir``, ``pw_shell``
373 (from ``struct_passwd``) and ``gr_name``, ``gr_gid`` (from ``struct_group``).
374 """
376 uid = uid or os.getuid()
377 u = pwd.getpwuid(uid)
379 r = dict(
380 pw_name=u.pw_name,
381 pw_uid=u.pw_uid,
382 pw_gid=u.pw_gid,
383 pw_dir=u.pw_dir,
384 pw_shell=u.pw_shell,
385 )
387 gid = gid or r['pw_gid']
388 g = grp.getgrgid(gid)
390 r['gr_name'] = g.gr_name
391 r['gr_gid'] = g.gr_gid
393 return r
396def find_entries(dirname: _Path, deep: bool = True):
397 """Find entries in a directory, skipping hidden ones.
399 Args:
400 dirname: Path to a directory.
401 deep: If true, also search subdirectories recursively.
403 Yields:
404 ``os.DirEntry`` objects for files and directories whose names do not start with a dot.
405 """
406 de: os.DirEntry
407 for de in os.scandir(dirname):
408 if de.name.startswith('.'):
409 continue
410 yield de
411 if de.is_dir() and deep:
412 yield from find_entries(de.path, deep=deep)
415def find_files(dirname: _Path, pattern=None, ext=None, deep: bool = True):
416 """Find files in a directory, skipping hidden ones.
418 Args:
419 dirname: Path to a directory.
420 pattern: Regular expression to search for in the file path.
421 ext: Extension or a list of extensions to match. Used only if ``pattern`` is not given.
422 deep: If true, also search subdirectories recursively.
424 Yields:
425 Paths of matching files.
426 """
427 if not pattern and ext:
428 if isinstance(ext, (list, tuple)):
429 ext = '|'.join(ext)
430 pattern = '\\.(' + ext + ')$'
432 for de in find_entries(dirname, deep=deep):
433 if de.is_file() and (pattern is None or re.search(pattern, de.path)):
434 yield de.path
437def find_directories(dirname: _Path, pattern=None, deep: bool = True):
438 """Find directories in a directory, skipping hidden ones.
440 Args:
441 dirname: Path to a directory.
442 pattern: Regular expression to search for in the directory path.
443 deep: If true, also search subdirectories recursively.
445 Yields:
446 Paths of matching directories.
447 """
449 for de in find_entries(dirname, deep=deep):
450 if de.is_dir() and (pattern is None or re.search(pattern, de.path)):
451 yield de.path
454class ParsePathResult(gws.Data):
455 """Components of a file path, as returned by ``parse_path``."""
457 path: str
458 """The full path."""
459 dirname: str
460 """The directory part."""
461 filename: str
462 """The file name, including the extension."""
463 stem: str
464 """The file name up to the first dot."""
465 extension: str
466 """The file name after the first dot, without the dot."""
469def parse_path(path: _Path) -> ParsePathResult:
470 """Split a file path into its components.
472 The extension is everything after the first dot in the file name, so ``a.tar.gz``
473 has the stem ``a`` and the extension ``tar.gz``. A file name starting with a dot has no extension.
475 Args:
476 path: File path.
478 Returns:
479 The path components.
480 """
482 str_path = _to_str(path)
483 sp = os.path.split(str_path)
485 pp = ParsePathResult(
486 path=str_path,
487 dirname='',
488 filename='',
489 stem='',
490 extension='',
491 )
492 pp.dirname = sp[0]
493 pp.filename = sp[1]
495 if pp.filename.startswith('.'):
496 pp.stem = pp.filename
497 else:
498 pp.stem, _, pp.extension = pp.filename.partition('.')
500 return pp
503def file_name(path: _Path) -> str:
504 """Return the file name part of a path.
506 Args:
507 path: File path.
509 Returns:
510 The last component of the path.
511 """
513 sp = os.path.split(_to_str(path))
514 return sp[1]
517def is_abs_path(path: _Path) -> bool:
518 """Check if a path is absolute.
520 Args:
521 path: File path.
523 Returns:
524 ``True`` if the path is absolute.
525 """
526 return os.path.isabs(path)
529def abs_path(path: _Path, base: _Path) -> str:
530 """Make a relative path absolute with respect to a base directory or file path.
532 If ``base`` is a file, its directory is used. An absolute ``path`` is only normalized.
534 Args:
535 path: A path.
536 base: Base directory or file path.
538 Returns:
539 The absolute path.
541 Raises:
542 ``ValueError``: If ``path`` is relative and ``base`` is empty.
543 """
545 str_path = _to_str(path)
547 if os.path.isabs(str_path):
548 return os.path.normpath(str_path)
550 if not base:
551 raise ValueError('cannot compute abspath without a base')
553 if os.path.isfile(base):
554 base = os.path.dirname(base)
556 return os.path.abspath(os.path.join(_to_str(base), str_path))
559def abs_web_path(path: str, basedir: str) -> Optional[str]:
560 """Resolve a web path in a base directory.
562 The path components must consist of letters, digits, ``_`` and ``-``,
563 the file name can also contain lowercase extensions. This prevents path traversal.
565 Args:
566 path: Slash-separated path, as received from a web request.
567 basedir: Path to the base directory.
569 Returns:
570 The path of an existing file in the base directory, or ``None`` if the path is invalid or the file does not exist.
571 """
573 _dir_re = r'^[A-Za-z0-9_-]+$'
574 _fil_re = r'^[A-Za-z0-9_-]+(\.[a-z0-9]+)*$'
576 gws.log.debug(f'abs_web_path: trying {path!r} in {basedir!r}')
578 dirs = []
579 for s in path.split('/'):
580 s = s.strip()
581 if s:
582 dirs.append(s)
584 if not dirs:
585 gws.log.warning(f'abs_web_path: empty path={path!r}')
586 return
588 fname = dirs.pop()
590 if not all(re.match(_dir_re, p) for p in dirs):
591 gws.log.warning(f'abs_web_path: invalid dirname in path={path!r}')
592 return
594 if not re.match(_fil_re, fname):
595 gws.log.warning(f'abs_web_path: invalid filename in path={path!r}')
596 return
598 p = basedir
599 if dirs:
600 p += '/' + '/'.join(dirs)
601 p += '/' + fname
603 if not os.path.isfile(p):
604 gws.log.warning(f'abs_web_path: not a file path={path!r}')
605 return
607 return p
610def rel_path(path: _Path, base: _Path) -> str:
611 """Make a path relative to a base directory or file path.
613 If ``base`` is a file, its directory is used.
615 Args:
616 path: Path to make relative.
617 base: Base directory or file path.
619 Returns:
620 The relative path.
621 """
623 if os.path.isfile(base):
624 base = os.path.dirname(base)
626 return os.path.relpath(_to_str(path), _to_str(base))
629def _to_str(p: _Path) -> str:
630 return p if isinstance(p, str) else bytes(p).decode('utf8')
633def _to_bytes(p: _Path) -> bytes:
634 return p if isinstance(p, bytes) else str(p).encode('utf8')