Coverage for gws-app/gws/core/util.py: 54%

1"""Core utilities 

2 

3Most common function which are needed everywhere. 

4 

5This module is available as ``gws.u`` everywhere. 

6""" 

7 

8import hashlib 

9import json 

10import os 

11import pickle 

12import random 

13import re 

14import shutil 

15import sys 

16import threading 

17import time 

18import urllib.parse 

19from typing import Optional, TypeVar, Union, cast 

20 

21from . import const, log 

22 

23 

24def is_data_object(x) -> bool: 

25 return False 

26 

27 

28def to_data_object(x): 

29 pass 

30 

31 

32def exit(code: int = 255): 

33 """Exit the application. 

34 

35 Args: 

36 code: Exit code. 

37 """ 

38 

39 sys.exit(code) 

40 

41 

42T = TypeVar('T') 

43 

44 

45def require(value: Optional[T], message: str = '') -> T: 

46 """Return the value if not None, otherwise raise an Exception.""" 

47 if value is None: 

48 raise ValueError(message or 'unexpected None value') 

49 return value 

50 

51 

52## 

53 

54# @TODO use ABC 

55 

56 

57def is_list(x): 

58 return isinstance(x, (list, tuple)) 

59 

60 

61def is_dict(x): 

62 return isinstance(x, dict) 

63 

64 

65def is_bytes(x): 

66 return isinstance(x, (bytes, bytearray)) 

67 # @TODO how to handle bytes-alikes? 

68 # return hasattr(x, 'decode') 

69 

70 

71def is_atom(x): 

72 return x is None or isinstance(x, (int, float, bool, str, bytes)) 

73 

74 

75def is_empty(x) -> bool: 

76 """Check if the value is empty (None, empty list/dict/object).""" 

77 

78 if x is None: 

79 return True 

80 try: 

81 return len(x) == 0 

82 except TypeError: 

83 pass 

84 try: 

85 return not vars(x) 

86 except TypeError: 

87 pass 

88 return False 

89 

90 

91## 

92 

93 

94def get(x, key, default=None): 

95 """Get a nested value/attribute from a structure. 

96 

97 Args: 

98 x: A dict, list or Data. 

99 key: A list or a dot separated string of nested keys. 

100 default: The default value. 

101 

102 Returns: 

103 The value if it exists and the default otherwise. 

104 """ 

105 

106 if not x: 

107 return default 

108 if isinstance(key, str): 

109 key = key.split('.') 

110 try: 

111 return _get(x, key) 

112 except (KeyError, IndexError, AttributeError, ValueError): 

113 return default 

114 

115 

116def has(x, key) -> bool: 

117 """True if a nested value/attribute exists in a structure. 

118 

119 Args: 

120 x: A dict, list or Data. 

121 key: A list or a dot separated string of nested keys. 

122 

123 Returns: 

124 True if a key exists 

125 """ 

126 

127 if not x: 

128 return False 

129 if isinstance(key, str): 

130 key = key.split('.') 

131 try: 

132 _get(x, key) 

133 return True 

134 except (KeyError, IndexError, AttributeError, ValueError): 

135 return False 

136 

137 

138def _get(x, keys): 

139 for k in keys: 

140 if is_dict(x): 

141 x = x[k] 

142 elif is_list(x): 

143 x = x[int(k)] 

144 elif is_data_object(x): 

145 # special case: raise a KeyError if the attribute is truly missing in a Data 

146 # (and not just equals to None) 

147 x = vars(x)[k] 

148 else: 

149 x = getattr(x, k) 

150 return x 

151 

152 

153def pop(x, key, default=None): 

154 if is_dict(x): 

155 return x.pop(key, default) 

156 if is_data_object(x): 

157 return vars(x).pop(key, default) 

158 return default 

159 

160 

161def pick(x, *keys): 

162 def _pick(d): 

163 r = {} 

164 for k in keys: 

165 if k in d: 

166 r[k] = d[k] 

167 return r 

168 

169 if is_dict(x): 

170 return _pick(x) 

171 if is_data_object(x): 

172 return type(x)(_pick(vars(x))) 

173 return {} 

174 

175 

176def omit(x, *keys): 

177 def _omit(d): 

178 r = {} 

179 for k, v in d.items(): 

180 if k not in keys: 

181 r[k] = d[k] 

182 return r 

183 

184 if is_dict(x): 

185 return _omit(x) 

186 if is_data_object(x): 

187 return type(x)(_omit(vars(x))) 

188 return {} 

189 

190 

191def collect(pairs): 

192 m = {} 

193 

194 for key, val in pairs: 

195 if key is not None: 

196 m.setdefault(key, []).append(val) 

197 

198 return m 

199 

200 

201def first(it): 

202 for x in it: 

203 return x 

204 

205 

206def first_not_none(*args): 

207 for a in args: 

208 if a is not None: 

209 return a 

210 

211 

212def merge(*args, **kwargs) -> Union[dict, 'Data']: 

213 """Create a new dict/Data object by merging values from dicts/Datas or kwargs. 

214 Latter vales overwrite former ones unless None. 

215 

216 Args: 

217 *args: dicts or Datas. 

218 **kwargs: Keyword args. 

219 

220 Returns: 

221 A new object (dict or Data). 

222 """ 

223 

224 def _merge(arg): 

225 for k, v in to_dict(arg).items(): 

226 if v is not None: 

227 m[k] = v 

228 

229 m = {} 

230 

231 for a in args: 

232 if a: 

233 _merge(a) 

234 if kwargs: 

235 _merge(kwargs) 

236 

237 if not args or isinstance(args[0], dict) or args[0] is None: 

238 return m 

239 return type(args[0])(m) 

240 

241 

242def deep_merge(x, y, concat_lists=True): 

243 """Deeply merge dicts/Datas into a nested dict/Data. 

244 Latter vales overwrite former ones unless None. 

245 

246 Args: 

247 x: dict or Data. 

248 y: dict or Data. 

249 concat_lists: if true, list will be concatenated, otherwise merged 

250 

251 Returns: 

252 A new object (dict or Data). 

253 """ 

254 

255 if (is_dict(x) or is_data_object(x)) and (is_dict(y) or is_data_object(y)): 

256 xd = to_dict(x) 

257 yd = to_dict(y) 

258 d = {k: deep_merge(xd.get(k), yd.get(k), concat_lists) for k in xd.keys() | yd.keys()} 

259 return d if is_dict(x) else type(x)(d) 

260 

261 if is_list(x) and is_list(y): 

262 xc = compact(x) 

263 yc = compact(y) 

264 if concat_lists: 

265 return xc + yc 

266 return [deep_merge(x1, y1, concat_lists) for x1, y1 in zip(xc, yc)] 

267 

268 return y if y is not None else x 

269 

270 

271def compact(x): 

272 """Remove all None values from a collection.""" 

273 

274 if is_dict(x): 

275 return {k: v for k, v in x.items() if v is not None} 

276 if is_data_object(x): 

277 d = {k: v for k, v in vars(x).items() if v is not None} 

278 return type(x)(d) 

279 return [v for v in x if v is not None] 

280 

281 

282def strip(x): 

283 """Strip all strings and remove empty values from a collection.""" 

284 

285 def _strip(v): 

286 if isinstance(v, (str, bytes, bytearray)): 

287 return v.strip() 

288 return v 

289 

290 def _dict(x1): 

291 d = {} 

292 for k, v in x1.items(): 

293 v = _strip(v) 

294 if not is_empty(v): 

295 d[k] = v 

296 return d 

297 

298 if is_dict(x): 

299 return _dict(x) 

300 if is_data_object(x): 

301 return type(x)(_dict(vars(x))) 

302 

303 r = [_strip(v) for v in x] 

304 return [v for v in r if not is_empty(v)] 

305 

306 

307def uniq(x): 

308 """Remove duplicate elements from a collection.""" 

309 

310 s = set() 

311 r = [] 

312 

313 for y in x: 

314 try: 

315 if y not in s: 

316 s.add(y) 

317 r.append(y) 

318 except TypeError: 

319 if y not in r: 

320 r.append(y) 

321 

322 return r 

323 

324 

325## 

326 

327 

328def to_int(x) -> int: 

329 """Convert a value to an int or 0 if this fails.""" 

330 

331 try: 

332 return int(x) 

333 except: 

334 return 0 

335 

336 

337def to_rounded_int(x) -> int: 

338 """Round and convert a value to an int or 0 if this fails.""" 

339 

340 try: 

341 if isinstance(x, float): 

342 return int(round(x)) 

343 return int(x) 

344 except: 

345 return 0 

346 

347 

348def to_float(x) -> float: 

349 """Convert a value to a float or 0.0 if this fails.""" 

350 

351 try: 

352 return float(x) 

353 except: 

354 return 0.0 

355 

356 

357def to_str(x, encodings: list[str] = None) -> str: 

358 """Convert a value to a string. 

359 

360 Args: 

361 x: Value. 

362 encodings: A list of acceptable encodings. If the value is bytes, try each encoding, 

363 and return the first one which passes without errors. 

364 

365 Returns: 

366 A string. 

367 """ 

368 

369 if isinstance(x, str): 

370 return x 

371 if x is None: 

372 return '' 

373 if not is_bytes(x): 

374 return str(x) 

375 if encodings: 

376 for enc in encodings: 

377 try: 

378 return x.decode(encoding=enc, errors='strict') 

379 except UnicodeDecodeError: 

380 pass 

381 return x.decode(encoding='utf-8', errors='ignore') 

382 

383 

384def to_bytes(x, encoding='utf8') -> bytes: 

385 """Convert a value to bytes by converting it to string and encoding.""" 

386 

387 if is_bytes(x): 

388 return bytes(x) 

389 if x is None: 

390 return b'' 

391 if not isinstance(x, str): 

392 x = str(x) 

393 return x.encode(encoding or 'utf8') 

394 

395 

396def to_list(x, delimiter: str = ',') -> list: 

397 """Convert a value to a list. 

398 

399 Args: 

400 x: A value. Is it's a string, split it by the delimiter 

401 delimiter: 

402 

403 Returns: 

404 A list. 

405 """ 

406 

407 if isinstance(x, list): 

408 return x 

409 if is_empty(x): 

410 return [] 

411 if is_bytes(x): 

412 x = to_str(x) 

413 if isinstance(x, str): 

414 if delimiter: 

415 ls = [s.strip() for s in x.split(delimiter)] 

416 return [s for s in ls if s] 

417 return [x] 

418 if isinstance(x, (int, float, bool)): 

419 return [x] 

420 try: 

421 return [s for s in x] 

422 except TypeError: 

423 return [] 

424 

425 

426def to_dict(x) -> dict: 

427 """Convert a value to a dict. If the argument is an object, return its `dict`.""" 

428 

429 if is_dict(x): 

430 return x 

431 if x is None: 

432 return {} 

433 try: 

434 f = getattr(x, '_asdict', None) 

435 if f: 

436 return f() 

437 return vars(x) 

438 except TypeError: 

439 raise ValueError(f'cannot convert {x!r} to dict') 

440 

441 

442def to_json_value(x) -> Union[dict, list, str, int, float, bool, None]: 

443 """Recursively convert a value to a JSON serializable type.""" 

444 

445 if is_atom(x): 

446 return x 

447 if is_dict(x): 

448 return {k: to_json_value(v) for k, v in x.items()} 

449 if is_list(x): 

450 return [to_json_value(v) for v in x] 

451 if is_data_object(x): 

452 return {k: to_json_value(v) for k, v in vars(x).items()} 

453 return str(x) 

454 

455 

456def to_upper_dict(x) -> dict: 

457 x = to_dict(x) 

458 return {k.upper(): v for k, v in x.items()} 

459 

460 

461def to_lower_dict(x) -> dict: 

462 x = to_dict(x) 

463 return {k.lower(): v for k, v in x.items()} 

464 

465 

466## 

467 

468_UID_DE_TRANS = { 

469 ord('ä'): 'ae', 

470 ord('ö'): 'oe', 

471 ord('ü'): 'ue', 

472 ord('ß'): 'ss', 

473} 

474 

475 

476def to_uid(x) -> str: 

477 """Convert a value to an uid (alphanumeric string).""" 

478 

479 if not x: 

480 return '' 

481 x = to_str(x).lower().strip().translate(_UID_DE_TRANS) 

482 x = re.sub(r'[^a-z0-9]+', '_', x) 

483 return x.strip('_') 

484 

485 

486def to_lines(txt: str, comment: str = None) -> list[str]: 

487 """Convert a multiline string into a list of strings. 

488 

489 Strip each line, skip empty lines, if `comment` is given, also remove lines starting with it. 

490 """ 

491 

492 ls = [] 

493 

494 for s in txt.splitlines(): 

495 if comment and comment in s: 

496 s = s.split(comment)[0] 

497 s = s.strip() 

498 if s: 

499 ls.append(s) 

500 

501 return ls 

502 

503 

504## 

505 

506 

507def parse_acl(acl): 

508 """Parse an ACL config into an ACL. 

509 

510 Args: 

511 acl: an ACL config. Can be given as a string ``allow X, allow Y, deny Z``, 

512 or as a list of dicts ``{ role X type allow }, { role Y type deny }``, 

513 or it can already be an ACL ``[1 X], [0 Y]``, 

514 or it can be None. 

515 

516 Returns: 

517 Access list. 

518 """ 

519 

520 if not acl: 

521 return [] 

522 

523 a = 'allow' 

524 d = 'deny' 

525 bits = {const.ALLOW, const.DENY} 

526 err = 'invalid ACL' 

527 

528 access = [] 

529 

530 if isinstance(acl, str): 

531 for p in acl.strip().split(','): 

532 s = p.strip().split() 

533 if len(s) != 2: 

534 raise ValueError(err) 

535 if s[0] == a: 

536 access.append((const.ALLOW, s[1])) 

537 elif s[0] == d: 

538 access.append((const.DENY, s[1])) 

539 else: 

540 raise ValueError(err) 

541 return access 

542 

543 if not isinstance(acl, list): 

544 raise ValueError(err) 

545 

546 if isinstance(acl[0], (list, tuple)): 

547 try: 

548 if all(len(s) == 2 and s[0] in bits for s in acl): 

549 return acl 

550 except (TypeError, IndexError): 

551 pass 

552 raise ValueError(err) 

553 

554 if isinstance(acl[0], dict): 

555 for s in acl: 

556 tk = s.get('type', '') 

557 rk = s.get('role', '') 

558 if not isinstance(rk, str): 

559 raise ValueError(err) 

560 if tk == a: 

561 access.append((const.ALLOW, rk)) 

562 elif tk == d: 

563 access.append((const.DENY, rk)) 

564 else: 

565 raise ValueError(err) 

566 return access 

567 

568 raise ValueError(err) 

569 

570 

571## 

572 

573UID_DELIMITER = '::' 

574 

575 

576def join_uid(parent_uid, object_uid): 

577 p = parent_uid.split(UID_DELIMITER) 

578 u = object_uid.split(UID_DELIMITER) 

579 return p[-1] + UID_DELIMITER + u[-1] 

580 

581 

582def split_uid(joined_uid: str) -> tuple[str, str]: 

583 p, _, u = joined_uid.partition(UID_DELIMITER) 

584 return p, u 

585 

586 

587## 

588 

589 

590def is_file(path): 

591 return os.path.isfile(path) 

592 

593 

594def is_dir(path): 

595 return os.path.isdir(path) 

596 

597 

598def read_file(path: str) -> str: 

599 try: 

600 with open(path, 'rt', encoding='utf8') as fp: 

601 return fp.read() 

602 except Exception as exc: 

603 log.debug(f'error reading {path=} {exc=}') 

604 raise 

605 

606 

607def read_file_b(path: str) -> bytes: 

608 try: 

609 with open(path, 'rb') as fp: 

610 return fp.read() 

611 except Exception as exc: 

612 log.debug(f'error reading {path=} {exc=}') 

613 raise 

614 

615 

616def write_file(path: str, s: str, user: int = None, group: int = None): 

617 try: 

618 with open(path, 'wt', encoding='utf8') as fp: 

619 fp.write(s) 

620 chown_default(path, user, group) 

621 return path 

622 except Exception as exc: 

623 log.debug(f'error writing {path=} {exc=}') 

624 raise 

625 

626 

627def write_file_b(path: str, s: str | bytes, user: int = None, group: int = None): 

628 if isinstance(s, str): 

629 s = s.encode('utf8') 

630 try: 

631 with open(path, 'wb') as fp: 

632 fp.write(s) 

633 chown_default(path, user, group) 

634 return path 

635 except Exception as exc: 

636 log.debug(f'error writing {path=} {exc=}') 

637 raise 

638 

639 

640def write_debug_file(path: str, s: str | bytes): 

641 """Write a debug file with the given content.""" 

642 

643 if isinstance(s, str): 

644 s = s.encode('utf8') 

645 try: 

646 d = ensure_dir(f'{const.VAR_DIR}/debug') 

647 with open(f'{d}/{path}', 'wb') as fp: 

648 fp.write(s) 

649 except Exception as exc: 

650 log.debug(f'error writing debug {path=} {exc=}') 

651 

652 

653def dirname(path): 

654 return os.path.dirname(path) 

655 

656 

657def ensure_dir(dir_path: str, base_dir: str = None, mode: int = 0o755, user: int = None, group: int = None) -> str: 

658 """Check if a (possibly nested) directory exists and create if it does not. 

659 

660 Args: 

661 dir_path: Path to a directory. 

662 base_dir: Base directory. 

663 mode: Directory creation mode. 

664 user: Directory user (defaults to gws.c.UID) 

665 group: Directory group (defaults to gws.c.GID) 

666 

667 Returns: 

668 The absolute path to the directory. 

669 """ 

670 

671 if base_dir: 

672 if os.path.isabs(dir_path): 

673 raise ValueError(f'cannot use an absolute path {dir_path!r} with a base dir') 

674 bpath = cast(bytes, os.path.join(base_dir.encode('utf8'), dir_path.encode('utf8'))) 

675 else: 

676 if not os.path.isabs(dir_path): 

677 raise ValueError(f'cannot use a relative path {dir_path!r} without a base dir') 

678 bpath = dir_path.encode('utf8') 

679 

680 if os.path.isdir(bpath): 

681 return bpath.decode('utf8') 

682 

683 parts = [] 

684 

685 for p in bpath.split(b'/'): 

686 parts.append(p) 

687 path = b'/'.join(parts) 

688 if path and not os.path.isdir(path): 

689 os.mkdir(path, mode) 

690 

691 chown_default(bpath, user, group) 

692 return bpath.decode('utf8') 

693 

694 

695def ensure_system_dirs(): 

696 for d in const.ALL_DIRS: 

697 ensure_dir(d) 

698 

699 

700def chown_default(path, user=None, group=None): 

701 try: 

702 os.chown(path, user or const.UID, group or const.GID) 

703 except OSError: 

704 pass 

705 

706 

707_ephemeral_state = dict( 

708 last_check_time=0, 

709 check_interval=2 * 3600, 

710 max_age=2 * 3600, 

711) 

712 

713 

714def ephemeral_path(name: str) -> str: 

715 """Return a new ephemeral path name.""" 

716 

717 if stime() > _ephemeral_state['last_check_time'] + _ephemeral_state['check_interval']: 

718 ephemeral_cleanup() 

719 

720 name = str(os.getpid()) + '_' + random_string(64) + '_' + name 

721 return const.EPHEMERAL_DIR + '/' + name 

722 

723 

724def ephemeral_dir(name: str) -> str: 

725 """Create and return an ephemeral directory.""" 

726 

727 return ensure_dir(const.EPHEMERAL_DIR + '/' + name) 

728 

729def ephemeral_set_max_age(seconds: int): 

730 """Set the maximum age for ephemeral paths.""" 

731 

732 _ephemeral_state['max_age'] = seconds 

733 

734 

735def ephemeral_cleanup(): 

736 """Remove ephemeral paths older than max age.""" 

737 

738 cnt = 0 

739 ts = stime() 

740 

741 for de in os.scandir(const.EPHEMERAL_DIR): 

742 try: 

743 age = int(ts - de.stat().st_mtime) 

744 if age > _ephemeral_state['max_age']: 

745 if de.is_dir(): 

746 shutil.rmtree(de.path) 

747 else: 

748 os.unlink(de.path) 

749 cnt += 1 

750 except (OSError, FileNotFoundError): 

751 pass 

752 

753 _ephemeral_state['last_check_time'] = ts 

754 

755 if cnt > 0: 

756 log.debug(f'ephemeral_cleanup: {cnt}') 

757 

758 

759def random_string(size: int) -> str: 

760 """Generate a random string of length `size`.""" 

761 

762 a = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789' 

763 r = random.SystemRandom() 

764 return ''.join(r.choice(a) for _ in range(size)) 

765 

766 

767class _FormatMapDefault: 

768 def __init__(self, d, default): 

769 self.d = d 

770 self.default = default 

771 

772 def __getitem__(self, item): 

773 val = self.d.get(item) 

774 return val if val is not None else self.default 

775 

776 

777def format_map(fmt: str, x: Union[dict, 'Data'], default: str = '') -> str: 

778 return fmt.format_map(_FormatMapDefault(x, default)) 

779 

780 

781def sha256(x): 

782 def _bytes(x): 

783 if is_bytes(x): 

784 return bytes(x) 

785 if isinstance(x, (int, float, bool)): 

786 return str(x).encode('utf8') 

787 if isinstance(x, str): 

788 return x.encode('utf8') 

789 

790 def _default(x): 

791 if is_data_object(x): 

792 return vars(x) 

793 return str(x) 

794 

795 c = _bytes(x) 

796 if c is None: 

797 j = json.dumps(x, default=_default, sort_keys=True, ensure_ascii=True) 

798 c = j.encode('utf8') 

799 

800 return hashlib.sha256(c).hexdigest() 

801 

802 

803class cached_property: 

804 """Decorator for a cached property.""" 

805 

806 def __init__(self, fn): 

807 self._fn = fn 

808 self.__doc__ = getattr(fn, '__doc__') 

809 

810 def __get__(self, obj, objtype=None): 

811 value = self._fn(obj) 

812 setattr(obj, self._fn.__name__, value) 

813 return value 

814 

815 

816# application lock/globals are global to one application 

817# server locks lock the whole server 

818# server globals are pickled in /tmp 

819 

820 

821_app_lock = threading.RLock() 

822 

823 

824def app_lock(name=''): 

825 return _app_lock 

826 

827 

828_app_globals: dict = {} 

829 

830 

831def get_app_global(name, init_fn): 

832 if name in _app_globals: 

833 return _app_globals[name] 

834 

835 with app_lock(name): 

836 if name not in _app_globals: 

837 _app_globals[name] = init_fn() 

838 

839 return _app_globals[name] 

840 

841 

842def set_app_global(name, value): 

843 with app_lock(name): 

844 _app_globals[name] = value 

845 return _app_globals[name] 

846 

847 

848def delete_app_global(name): 

849 with app_lock(name): 

850 _app_globals.pop(name, None)