Load a dirty app class from an import path. Args: import_path: String in format 'module.path:ClassName' Returns: An instance of the dirty app class Raises: DirtyAppNotFoundError: If the module or class cannot be found DirtyAppError: If the class is
(import_path)
| 213 | |
| 214 | |
| 215 | def load_dirty_app(import_path): |
| 216 | """ |
| 217 | Load a dirty app class from an import path. |
| 218 | |
| 219 | Args: |
| 220 | import_path: String in format 'module.path:ClassName' |
| 221 | |
| 222 | Returns: |
| 223 | An instance of the dirty app class |
| 224 | |
| 225 | Raises: |
| 226 | DirtyAppNotFoundError: If the module or class cannot be found |
| 227 | DirtyAppError: If the class is not a valid DirtyApp subclass |
| 228 | """ |
| 229 | if ':' not in import_path: |
| 230 | raise DirtyAppError( |
| 231 | f"Invalid import path format: {import_path}. " |
| 232 | f"Expected 'module.path:ClassName'", |
| 233 | app_path=import_path |
| 234 | ) |
| 235 | |
| 236 | module_path, class_name = import_path.rsplit(':', 1) |
| 237 | |
| 238 | try: |
| 239 | # Import the module |
| 240 | if module_path in sys.modules: |
| 241 | module = sys.modules[module_path] |
| 242 | else: |
| 243 | module = importlib.import_module(module_path) |
| 244 | except ImportError as e: |
| 245 | raise DirtyAppNotFoundError(import_path) from e |
| 246 | |
| 247 | # Get the class from the module |
| 248 | try: |
| 249 | app_class = getattr(module, class_name) |
| 250 | except AttributeError: |
| 251 | raise DirtyAppNotFoundError(import_path) from None |
| 252 | |
| 253 | # Validate it's a class |
| 254 | if not isinstance(app_class, type): |
| 255 | raise DirtyAppError( |
| 256 | f"{import_path} is not a class", |
| 257 | app_path=import_path |
| 258 | ) |
| 259 | |
| 260 | # Create an instance |
| 261 | try: |
| 262 | app = app_class() |
| 263 | except Exception as e: |
| 264 | raise DirtyAppError( |
| 265 | f"Failed to instantiate {import_path}: {e}", |
| 266 | app_path=import_path |
| 267 | ) from e |
| 268 | |
| 269 | # Validate it has the required methods |
| 270 | required_methods = ['init', '__call__', 'close'] |
| 271 | for method_name in required_methods: |
| 272 | if not hasattr(app, method_name) or not callable(getattr(app, method_name)): |