Coverage for databooks/common.py: 89%

1"""Common set of miscellaneous functions.""" 

2 

3from itertools import chain 

4from pathlib import Path 

5from typing import Collection, List, Optional, Sequence 

6 

7from databooks.logging import get_logger 

8 

9logger = get_logger(__file__) 

10 

11 

12def expand_paths( 

13 paths: List[Path], *, ignore: Sequence[str] = ("!*",), rglob: str = "*.ipynb" 

14) -> Optional[List[Path]]: 

15 """ 

16 Get paths of existing file from list of directory or file paths. 

17 

18 :param paths: Paths to consider (can be directories or files) 

19 :param ignore: Glob expressions of files to ignore 

20 :param rglob: Glob expression for expanding directory paths and filtering out 

21 existing file paths (i.e.: to retrieve only notebooks) 

22 :return: List of existing file paths 

23 """ 

24 if not paths: 

25 return None 

26 filepaths = set( 

27 chain.from_iterable( 

28 list(path.resolve().rglob(rglob)) if path.is_dir() else [path] 

29 for path in paths 

30 ) 

31 ) 

32 common_path = find_common_parent(paths=paths) 

33 ignored = set(chain.from_iterable(common_path.rglob(i) for i in ignore)) 

34 ignored = {p.resolve() for p in ignored} 

35 logger.debug( 

36 f"{len(ignored)} files will be ignored from {len(filepaths)} file paths." 

37 ) 

38 valid_filepaths = [p for p in filepaths - ignored if p.is_file()] 

39 

40 if not valid_filepaths: 

41 logger.debug( 

42 f"There are no files in {paths} (ignoring {ignore}) that match `{rglob}`." 

43 ) 

44 return valid_filepaths 

45 

46 

47def find_common_parent(paths: Collection[Path]) -> Path: 

48 """Find common parent amongst several file paths (includes current path).""" 

49 if not paths: 

50 raise ValueError(f"Expected non-empty `paths`, got {paths}.") 

51 return max(set.intersection(*[{*p.resolve().parents, p.resolve()} for p in paths])) 

52 

53 

54def find_obj( 

55 obj_name: str, start: Path, finish: Path, is_dir: bool = False 

56) -> Optional[Path]: 

57 """ 

58 Recursively find file along directory path, from the end (child) directory to start. 

59 

60 :param obj_name: File name to locate 

61 :param start: Start (parent) directory 

62 :param finish: Finish (child) path 

63 :param is_dir: Whether object is a directory or a file 

64 :return: File path 

65 """ 

66 finish = finish if finish.is_dir() else finish.parent 

67 logger.debug(f"Searching for {obj_name} between {start} and {finish}.") 

68 if not start.is_dir(): 

69 raise ValueError("Parameter `start` must be a directory.") 

70 

71 if start.resolve() not in [finish, *finish.resolve().parents]: 

72 logger.debug( 

73 f"Parameter `start` is not a parent directory of `finish` (for {start} and" 

74 f" {finish}). Cannot find {obj_name}." 

75 ) 

76 return None 

77 

78 is_obj = (finish / obj_name).is_dir() if is_dir else (finish / obj_name).is_file() 

79 if is_obj: 

80 return finish / obj_name 

81 elif finish.samefile(start): 

82 logger.debug(f"{obj_name} not found between {start} and {finish}.") 

83 return None 

84 else: 

85 return find_obj( 

86 obj_name=obj_name, start=start, finish=finish.parent, is_dir=is_dir 

87 )