1+ import importlib .util
2+ import inspect
3+ from pathlib import Path
4+ from typing import Any , Iterable , Type , Optional
5+
6+ from loguru import logger
7+
8+
9+ class ModuleImporter :
10+ """
11+ A class that handles dynamic module importing with caching capabilities.
12+ Provides functionality to import modules from file paths and extract classes from them.
13+ """
14+
15+ def __init__ (self ) -> None :
16+ # Module cache to prevent duplicate imports
17+ self ._module_cache : dict [str , Any ] = {}
18+
19+ def import_module_from_path (self , file_path : str ) -> Optional [Any ]:
20+ """
21+ Import a module from a file path with caching.
22+
23+ Args:
24+ file_path (str): The file path of the module to import.
25+
26+ Returns:
27+ Optional[Any]: The imported module or None if import fails.
28+ """
29+ resolved_path = Path (file_path ).resolve ()
30+ cache_key = str (resolved_path )
31+ module_name = resolved_path .stem
32+
33+ # Check if module is already cached
34+ if cache_key in self ._module_cache :
35+ logger .debug (f"[MODULE CACHE] Using cached module: { module_name } )
36+ return self ._module_cache [cache_key ]
37+
38+ logger .info (f"[MODULE IMPORT] Import module path: { resolved_path } )
39+
40+ # Create a module specification
41+ spec = importlib .util .spec_from_file_location (module_name , resolved_path )
42+ if spec is None :
43+ logger .warning (f"[MODULE IMPORT] Could not create spec for { module_name } )
44+ return None
45+
46+ # Create a new module based on the specification
47+ module = importlib .util .module_from_spec (spec )
48+ if spec .loader is None :
49+ logger .warning (f"[MODULE IMPORT] No loader found for { module_name } )
50+ return None
51+
52+ # Execute the module in its own namespace
53+ logger .info (f"[MODULE IMPORT] Import module: { module_name } )
54+ try :
55+ spec .loader .exec_module (module )
56+ logger .success (f"[MODULE IMPORT] Successfully imported { module_name } )
57+ # Cache the module
58+ self ._module_cache [cache_key ] = module
59+ return module
60+ except Exception as error :
61+ logger .warning (f"[MODULE IMPORT] Failed to import { module_name } { error } )
62+ return None
63+
64+ def extract_classes_from_module (self , module : Any ) -> list [Type [object ]]:
65+ """
66+ Extract all classes from a module.
67+
68+ Args:
69+ module (Any): The module to extract classes from.
70+
71+ Returns:
72+ list[Type[object]]: List of classes found in the module.
73+ """
74+ loaded_classes = []
75+ for attr in dir (module ):
76+ obj = getattr (module , attr )
77+ if attr .startswith ("__" ):
78+ continue
79+ if not inspect .isclass (obj ):
80+ continue
81+ loaded_classes .append (obj )
82+ return loaded_classes
83+
84+ def import_classes_from_paths (
85+ self ,
86+ file_paths : Iterable [str ],
87+ target_subclasses : Iterable [Type [object ]] = [],
88+ ignore_errors : bool = True
89+ ) -> set [Type [object ]]:
90+ """
91+ Import classes from multiple file paths with optional filtering.
92+
93+ Args:
94+ file_paths (Iterable[str]): The file paths of the modules to import.
95+ target_subclasses (Iterable[Type[object]], optional): Target subclasses to filter. Defaults to [].
96+ ignore_errors (bool, optional): Whether to ignore import errors. Defaults to True.
97+
98+ Returns:
99+ set[Type[object]]: Set of imported classes.
100+ """
101+ all_loaded_classes : list [Type [object ]] = []
102+
103+ for file_path in file_paths :
104+ module = self .import_module_from_path (file_path )
105+ if module is None :
106+ if not ignore_errors :
107+ raise ImportError (f"Failed to import module from { file_path } )
108+ continue
109+
110+ loaded_classes = self .extract_classes_from_module (module )
111+ all_loaded_classes .extend (loaded_classes )
112+
113+ returned_target_classes : set [Type [object ]] = set ()
114+
115+ # If no target subclasses specified, return all loaded classes
116+ if not target_subclasses :
117+ returned_target_classes = set (all_loaded_classes )
118+ else :
119+ # Filter classes based on target subclasses
120+ for target_cls in target_subclasses :
121+ for loaded_class in all_loaded_classes :
122+ if loaded_class in target_subclasses :
123+ continue
124+ if issubclass (loaded_class , target_cls ):
125+ returned_target_classes .add (loaded_class )
126+
127+ return returned_target_classes
128+
129+ def clear_cache (self ) -> None :
130+ """Clear the module cache. Useful for testing or when you need to force re-import."""
131+ self ._module_cache .clear ()
132+
133+ def get_cache_size (self ) -> int :
134+ """Get the number of cached modules."""
135+ return len (self ._module_cache )
0 commit comments