Coverage for gws-app/gws/lib/importer/__init__.py: 21%

1"""Handle dynamic imports""" 

2 

3import sys 

4import os 

5import importlib 

6 

7import gws 

8 

9 

10class Error(gws.Error): 

11 pass 

12 

13 

14def import_from_path(path, base_dir=gws.c.APP_DIR): 

15 abs_path = _abs_path(path, base_dir) 

16 if not os.path.isfile(abs_path): 

17 raise Error(f'{abs_path!r}: not found') 

18 

19 if abs_path.startswith(base_dir): 

20 # our own module, import relatively to base_dir 

21 return _do_import(abs_path, base_dir) 

22 

23 # plugin module, import relatively to the bottom-most "namespace" dir (without __init__) 

24 

25 dirs = abs_path.strip('/').split('/') 

26 dirs.pop() 

27 

28 for n in range(len(dirs), 0, -1): 

29 ns_dir = '/' + '/'.join(dirs[:n]) 

30 if not os.path.isfile(ns_dir + '/__init__.py'): 

31 return _do_import(abs_path, ns_dir) 

32 

33 raise Error(f'{abs_path!r}: cannot locate a base directory') 

34 

35 

36def _abs_path(path, base_dir): 

37 if not os.path.isabs(path): 

38 path = os.path.join(base_dir, path) 

39 path = os.path.normpath(path) 

40 if os.path.isdir(path): 

41 path += '/__init__.py' 

42 return path 

43 

44 

45def _do_import(abs_path, base_dir): 

46 mod_name = _module_name(abs_path[len(base_dir):]) 

47 

48 if mod_name in sys.modules: 

49 mpath = getattr(sys.modules[mod_name], '__file__', None) 

50 if mpath != abs_path: 

51 raise Error(f'{abs_path!r}: overwriting {mod_name!r} from {mpath!r}') 

52 return sys.modules[mod_name] 

53 

54 gws.log.debug(f'import: {abs_path=} {mod_name=} {base_dir=}') 

55 

56 if base_dir not in sys.path: 

57 sys.path.insert(0, base_dir) 

58 

59 try: 

60 return importlib.import_module(mod_name) 

61 except Exception as exc: 

62 raise Error(f'{abs_path!r}: import failed') from exc 

63 

64 

65def _module_name(path): 

66 parts = path.strip('/').split('/') 

67 if parts[-1] == '__init__.py': 

68 parts.pop() 

69 elif parts[-1].endswith('.py'): 

70 parts[-1] = parts[-1][:-3] 

71 return '.'.join(parts)