loader — PluginLoader

Overview

PluginLoader discovers plugin classes on disk and registers instances of them into a PluginRegistry. It supports eager discovery (discover_all), on-demand lazy loading (register_lazy / load_lazy), and the file-mtime bookkeeping that both its own reload_file() and the separate PluginHotReloader rely on.

PluginLoader

class PluginLoader:
    def __init__(
        self,
        registry: PluginRegistry,
        plugin_dirs: Optional[List[str]] = None,
        default_config: Optional[PluginConfig] = None,
        module_prefix: str = "rag_plugins",
    ) -> None: ...

Parameters

discover_all() -> List[str]

Scans every directory in self.plugin_dirs via _scan_directory, aggregates the list of successfully registered plugin names, and logs a summary. This is what PluginManager.startup() calls when auto_discover=True.

discover_directory(directory: str) -> List[str]

Discovers from one directory that is not necessarily in self.plugin_dirs — an ad-hoc scan.

register_lazy(name: str, file_path: str) -> None

Records name → Path(file_path) in an internal map without importing anything.

load_lazy(name: str) -> Optional[BasePlugin]

Looks up name in the lazy map, calls _load_file(path), and returns the instance whose .name matches (or None if the name wasn't registered lazily, or if loading it didn't produce a matching instance). Logs a warning and returns None if name was never registered via register_lazy.

get_changed_files() -> List[Path]

Compares each tracked file's current stat().st_mtime against the mtime recorded at last load; returns files that are newer. Files that no longer exist() are silently skipped (not reported as "changed").

reload_file(file_path: Path) -> List[str]

Internal workflow:

1. For every name currently tracked in self._loaded_names, look it up in the registry; if the plugin's type(plugin).__module__ contains file_path.stem, unregister it.
2. Pop the corresponding synthetic module name from sys.modules so Python re-executes the file on next import rather than returning a cached module.
3. Call _load_file(file_path) again and collect the newly instantiated names.
4. Log and return the list of reloaded names.

Side effects: mutates the registry (unregister + register), mutates sys.modules, mutates self._loaded_names and self._file_mtimes.

[!NOTE] Step 1's module-name matching is a substring check (file_path.stem in src), not an exact-module comparison. If two different plugin files happen to share a stem as a substring of each other's module path (unlikely in normal usage, since each file gets its own module_prefix.stem), this could match more broadly than intended. In the common case of one file → one distinct stem, this behaves as expected.

Internal scanning and instantiation

_scan_directory(directory: Path) -> List[str]

Returns [] with a warning log if the directory doesn't exist. Otherwise iterates sorted(directory.glob("*.py")), skipping files whose name starts with _ (so __init__.py and any _private.py helper files are never scanned), and calls _load_file on each.

Note: this is a non-recursive glob — subdirectories of a configured plugin_dirs entry are not scanned.

_load_file(path: Path) -> List[BasePlugin]

1. Builds a synthetic module name f"{module_prefix}.{path.stem}".
2. Uses importlib.util.spec_from_file_location + module_from_spec + spec.loader.exec_module to import the file as a standalone module, registering it in sys.modules under the synthetic name before execution (so the module can see itself in sys.modules if it needs to, and so recursive imports resolve).
3. Records the file's current mtime in self._file_mtimes.
4. Calls _find_plugin_classes(module) to discover concrete BasePlugin subclasses defined in that module (not imported into it).
5. Calls _instantiate(cls, path) for each.
6. Catches and logs (with traceback via exc_info=True) any exception during import, returning whatever instances were found before the failure (typically none, if the import itself failed).

Raises: Does not raise — all exceptions are caught internally and logged; callers get an empty or partial list back on failure rather than an exception.

_instantiate(cls, source: Path) -> Optional[BasePlugin]

Reads name = cls.METADATA.name (falling back to cls.__name__ if METADATA isn't set — though BasePlugin.__init__ will itself raise TypeError a moment later if that's the case). Skips instantiation (returns None, debug log) if a plugin with that name is already registered. Otherwise instantiates with default_config, registers it, tracks the name in _loaded_names, and returns the instance. Any exception during instantiation or registration is caught, logged with a traceback, and results in None.

_find_plugin_classes(module) -> List[Type[BasePlugin]] (module-private function)

Uses inspect.getmembers(module, inspect.isclass) and keeps a class only if all of:

• It is a subclass of BasePlugin.
• It is not BasePlugin itself.
• obj.__module__ == module.__name__ — i.e. the class was defined in this file, not merely imported into it (this prevents re-registering BasePlugin subclasses that a plugin file imports from elsewhere in the package).
• not inspect.isabstract(obj) — abstract classes (e.g. a plugin that hasn't implemented all abstract methods) are skipped.
• hasattr(obj, "METADATA").

Complexity

discover_all() / _scan_directory() is O(D × F) in the number of directories D and files F per directory, each involving a full Python import (dominant cost). get_changed_files() is O(T) in the number of currently tracked files, each requiring one stat() syscall.

Thread safety

Not thread-safe: sys.modules mutation and dict tracking (_file_mtimes, _loaded_names) are not guarded by any lock. PluginHotReloader (which calls reload_file from a background thread via asyncio.run(...)) is the one part of this codebase that exercises this path off the main thread — see the caution note in hot_reload.

Side effects

File I/O (reading .py files, stat() calls), dynamic module execution (arbitrary code from the loaded file runs at import time), mutation of sys.modules and the target PluginRegistry, structured logging.

Examples

1. Basic discovery:

from zerda.plugins import PluginRegistry, PluginLoader

registry = PluginRegistry()
loader = PluginLoader(registry, plugin_dirs=["./plugins/builtin", "./plugins/custom"])
names = await loader.discover_all()

2. Lazy loading:

loader.register_lazy("heavy_model_plugin", "/path/to/heavy_model_plugin.py")
# ... nothing imported yet ...
plugin = loader.load_lazy("heavy_model_plugin")   # imports & instantiates now

3. Manual hot-reload check:

changed = loader.get_changed_files()
for path in changed:
    reloaded_names = await loader.reload_file(path)

4. Error handling — bad directory:

names = await loader.discover_directory("/does/not/exist")
assert names == []   # logged a warning, returned empty rather than raising

5. Production usage inside a manager:

manager = PluginManager(plugin_dirs=["./plugins/prod"])
await manager.startup()   # internally: loader.discover_all(), then initialize each plugin

Best practices

• Keep one plugin class per file where practical — reload_file's substring-based module matching (see note above) is simplest to reason about with a 1:1 file-to-plugin mapping.
• Prefix any helper/non-plugin .py files in a plugin directory with _ so the scanner skips them.

Common mistakes

• Expecting discover_all() to recurse into subdirectories — it only globs the top level of each configured directory.
• Assuming _load_file raises on import errors — it swallows and logs them, returning an empty list, so check the return value / registry state rather than wrapping calls in try/except.

Related APIs

• PluginRegistry — receives loaded instances
• PluginManager.startup — the typical caller of discover_all
• PluginHotReloader — calls reload_file/get_changed_files from a background thread