PluginManager

PluginManager

PluginManager is the central class that orchestrates configuration, discovery, lifecycle, and hooks.

Constructor

PluginManager(
    config_path: str = "config/app.yaml",
    pre_activate: Callable[[BasePlugin, dict], bool] | None = None,
    api_version: str = "1",
)

Parameter	Description
config_path	Path to app.yaml configuration file
pre_activate	Optional callback (plugin, config) -> bool called before activation. Return False to reject a plugin.
api_version	Current hook spec version. Plugins with a different api_version will log a warning but still load.

On creation, the manager:

1. Loads the app config from config_path
2. Creates a pluggy.PluginManager with the entry_point_group from config
3. Initializes the lifecycle tracker
4. Initializes i18n with the default_language from config

Methods

Config

get_app_config() -> dict

Returns the loaded application configuration.

get_plugin_config(plugin_name: str) -> dict

Loads and returns the config for a specific plugin from config/plugins/{name}.yaml.

reload_config() -> None

Reload application config from disk. Reloads app.yaml and clears the i18n cache. Active plugins are not affected - call deactivate_all() + discover_plugins() to fully restart with new config.

pm.reload_config()

Discovery and Registration

list_available_plugins() -> list[str]

Return names of all discoverable plugins from entry points without loading them. Useful for settings UIs.

available = pm.list_available_plugins()
# ["export", "analytics", "grammar"]

discover_plugins() -> None

Full automatic pipeline:

1. Load plugins from entry points
2. Filter by enabled/disabled config
3. Check and skip plugins with missing dependencies
4. Topologically sort by dependencies
5. Init and activate each plugin

pm = PluginManager("config/app.yaml")
pm.discover_plugins()

register_plugins(plugin_classes: list[type[BasePlugin]]) -> None

Register plugin classes directly without entry point discovery. Useful for testing or programmatic registration. Applies the same filtering, dependency checking, and lifecycle as discover_plugins().

pm.register_plugins([HelloPlugin, ExportPlugin])

register_plugin(plugin: BasePlugin, plugin_config: dict | None = None) -> None

Register a single pre-instantiated plugin. Useful for tests or dynamically created plugins. If plugin_config is not provided, it is loaded from YAML.

plugin = HelloPlugin()
pm.register_plugin(plugin)

# Or with explicit config:
pm.register_plugin(plugin, plugin_config={"greeting": "Hi"})

Lifecycle

activate_plugin(name: str) -> None

Activate a specific initialized plugin by name.

deactivate_plugin(name: str) -> None

Deactivate a specific active plugin and unregister its hooks from pluggy. After deactivation, the plugin's @hookimpl methods are no longer called.

deactivate_all() -> None

Deactivate all active plugins in reverse activation order (LIFO) and unregister hooks.

reload_plugin(name: str) -> bool

Hot-reload a plugin: deactivate, re-import module, re-init, activate. Code changes on disk take effect without restarting the application.

success = pm.reload_plugin("export")

Returns True on success, False on failure (check get_load_errors() for details).

get_plugin(name: str) -> BasePlugin | None

Get a plugin instance by name. Returns None if not found.

get_active_plugins() -> list[BasePlugin]

Return all currently active plugins.

Error Reporting

get_load_errors() -> dict[str, str]

Return errors from plugin loading/activation. Maps plugin name to error message. Tracks missing dependencies, init failures, activation failures, and pre-activate rejections.

errors = pm.get_load_errors()
# {"premium": "Rejected by pre-activate check", "broken": "Failed to initialize"}

Health Checks

health_check() -> dict[str, dict]

Run health checks on all active plugins. Each plugin's health() method is called. Exceptions are caught and reported.

status = pm.health_check()
# {"export": {"status": "ok"}, "grammar": {"status": "error", "error": "API unreachable"}}

Hooks

register_hookspecs(spec_module: object) -> None

Register hook specifications from a module containing @hookspec-decorated functions.

pm.register_hookspecs(my_hookspecs)

call_hook(hook_name: str, **kwargs) -> list

Call a named hook on all registered plugins. Returns a list of results. Logs a warning if the hook is not found. If any implementation throws, returns [].

results = pm.call_hook("on_document_save", document=doc)

call_hook_safe(hook_name: str, **kwargs) -> list

Call a hook, executing each implementation individually. Failed implementations are logged and skipped, others still execute. Recommended for non-critical hooks where partial results are acceptable.

results = pm.call_hook_safe("on_document_save", document=doc)

Introspection

get_plugin_hooks(name: str) -> list[str]

Return hook names implemented by a specific plugin. Useful for debugging and settings UIs.

hooks = pm.get_plugin_hooks("export")
# ["export_execute", "on_document_save"]

get_all_hook_names() -> list[str]

Return all registered hook spec names.

names = pm.get_all_hook_names()
# ["on_startup", "on_shutdown", "on_document_save"]

Extensions

get_extensions(extension_point: type) -> list[BasePlugin]

Return all active plugins that implement a given extension point (class or ABC). See Extensions for the full pattern.

formats = pm.get_extensions(ExportFormat)

FastAPI Integration

mount_routes(app: FastAPI, prefix: str = "/api") -> None

Mount routes from all active plugins onto a FastAPI application. The prefix parameter controls the URL prefix (default: /api). Plugins bring their own route prefixes via their routers.

from fastapi import FastAPI

app = FastAPI()
pm.mount_routes(app)
pm.mount_routes(app, prefix="/v2/api")  # custom prefix

i18n

get_text(key: str, lang: str | None = None) -> str

Get an internationalized string by dot-notation key. Falls back to default language if not found.

pm.get_text("common.save", "de")  # "Speichern"

Alembic Integration

collect_migrations() -> dict[str, str]

Collect Alembic migration directories from all active plugins. Returns a dict mapping plugin name to directory path.

Complete Usage

from pluginforge import PluginManager

# Setup with pre-activate callback (e.g. license check)
def check_license(plugin, config):
    if plugin.name in premium_plugins:
        return validate_license(plugin.name)
    return True

pm = PluginManager(
    "config/app.yaml",
    pre_activate=check_license,
    api_version="2",
)
pm.register_hookspecs(my_hooks)
pm.discover_plugins()

# Check for errors
for name, error in pm.get_load_errors().items():
    print(f"Plugin '{name}' failed: {error}")

# Use hooks
results = pm.call_hook("on_startup")

# Query extension points
exporters = pm.get_extensions(ExportFormat)

# Health check
status = pm.health_check()

# i18n
title = pm.get_text("app.title", "de")

# FastAPI (optional)
from fastapi import FastAPI
app = FastAPI()
pm.mount_routes(app)

# Hot-reload during development
pm.reload_plugin("export")

# Shutdown
pm.deactivate_all()