Skip to content

PluginManager

Jump to bottom
Asterios Raptis edited this page May 21, 2026 · 7 revisions

PluginManager

PluginManager is the central class that orchestrates configuration, discovery, lifecycle, hooks, (since v0.6.0) live config refresh and entry-point rediscovery, and (since v0.10.0) explicit config-overlay support via merge_app_config().

Constructor

PluginManager(
    config_path: str = "config/app.yaml",
    pre_activate: Callable[[BasePlugin, dict], bool] | None = None,
    api_version: str = "1",
    app_version: str | None = None,
    api_version_severity: Literal["warning", "error"] = "warning",
    app_version_severity: Literal["warning", "error"] = "error",
    app_id: str | None = None,
)
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 (filter reason: pre_activate_rejected).
api_version Current hook spec version. Plugins with a different api_version are gated by api_version_severity.
app_version Host application version. When set, plugins declaring min_app_version are gated against this minimum (v0.6.0). When None, min_app_version declarations are ignored.
api_version_severity "warning" (default) logs and proceeds on api_version mismatch; "error" filters with incompatible_api_version. (v0.6.0)
app_version_severity "error" (default) filters when app_version < min_app_version with incompatible_app_version; "warning" logs and proceeds. (v0.6.0)
app_id Host application identifier. When set, plugins declaring target_application are validated against this value; mismatches filter with wrong_application (v0.7.0), and plugins not declaring target_application filter with missing_target_application (v0.9.0). When None, plugin identity claims are not validated (back-compat).

On creation, the manager loads app.yaml, creates a pluggy.PluginManager with the configured entry_point_group, initializes the lifecycle tracker, and initializes i18n with the configured default_language.

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 app.yaml from disk and clear the i18n cache. Active plugins are not notified. For an in-place update that notifies plugins, use refresh_config() instead.

refresh_config(new_app_config: dict | None = None, *, notify: bool = True) -> list[PluginError] (v0.6.0; notify kwarg added v0.10.0)

Replace the app-config snapshot and (by default) notify each active plugin via the on_config_changed(old, new) hook. With no argument, reloads app.yaml from disk and refreshes i18n. With an explicit dict, replaces the snapshot directly.

Set notify=False to swap the snapshot without firing on_config_changed. Appropriate at startup before discover_plugins() when no plugins are active to notify, or when a caller is pushing a pre-computed snapshot and wants to manage notifications separately. The return value is always [] under notify=False.

Distinction from merge_app_config (v0.10.0): refresh_config REPLACES the entire snapshot. Pre-existing keys not present in new_app_config are LOST. Use merge_app_config for overlay semantics that preserve unmentioned keys.

Returns a list of PluginError entries from plugins whose on_config_changed raised. Failures do not stop other plugins from being notified.

errors = pm.refresh_config({"app": {"name": "Renamed"}, "plugins": {"enabled": ["export"]}})
for e in errors:
    print(f"{e.name}: {e.user_facing_message}")

See Lifecycle for the full notification flow.

merge_app_config(overlay: dict, *, notify: bool = True) -> list[PluginError] (v0.10.0)

Recursively merge overlay onto the current app-config snapshot. Merge semantics: dicts merge recursively; lists, scalars, and None replace. Pre-existing keys not mentioned in the overlay are preserved.

When notify=True (the default), each active plugin's on_config_changed(old, new) is called with the pre- and post-merge snapshots via the same codepath as refresh_config. When notify=False, the snapshot is updated silently - appropriate at startup before discover_plugins() when no plugins are active to notify.

# Pre-discovery overlay (startup, before discover_plugins):
pm.merge_app_config(user_overlay_dict, notify=False)
pm.discover_plugins()

# Post-discovery overlay (Settings UI write, env-var change):
errors = pm.merge_app_config(secrets_layer)  # notify=True default

Distinction from refresh_config: merge_app_config MERGES an overlay into the existing config; refresh_config REPLACES the entire config. The two methods are deliberate alternatives, not aliases.

The internal _deep_merge implementation is module-private and NOT exposed as a public utility. Consumers needing custom merge semantics (e.g. lists concatenate instead of replace) pre-compute the merged dict themselves and use refresh_config(merged_dict, notify=...).

Migration: v0.10.0 replaces the manager._app_config = merged # type: ignore[attr-defined] pattern used by some downstream consumers. The private-attribute write keeps working through v0.10.0 but is slated for removal at v1.0.0.

Discovery and Registration

list_available_plugins() -> list[str]

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

discover_plugins() -> DiscoveryResult (return type since v0.6.0)

Full automatic discovery pipeline:

  1. Load plugins from entry points
  2. Filter by enabled / disabled config
  3. Check identity (v0.7.0), api_version, app_version, dependencies
  4. Topologically sort by depends_on
  5. Init, run pre_activate, activate each plugin

Returns a DiscoveryResult with per-plugin state and any errors. Callers that ignored the return value pre-v0.6.0 continue to work.

result = pm.discover_plugins()
print(f"Activated: {result.activated}")
print(f"Filtered:  {result.filtered_out()}")  # {name: filter_reason}

register_plugins(plugin_classes: list[type[BasePlugin]]) -> DiscoveryResult (return type since v0.6.0)

Register plugin classes directly without entry-point discovery. Same filtering, dependency check, and lifecycle as discover_plugins().

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

Register a single pre-instantiated plugin. Useful for tests or dynamically created plugins. Applies the same identity / version / pre-activate gates inline.

rediscover() -> DiscoveryDiff (v0.6.0)

Re-read entry points after invalidating importlib and importlib.metadata caches, and reconcile against currently active plugins. Activates newly-discovered plugins, deactivates removed ones, leaves unchanged ones untouched. Does NOT reload module code for unchanged plugins; use reload_plugin(name) for that.

diff = pm.rediscover()
print(f"Added:     {diff.added}")
print(f"Removed:   {diff.removed}")
print(f"Unchanged: {diff.unchanged}")

Canonical use case: a new plugin distribution was installed in another shell (poetry install, pip install) while the application was running, and the host wants to pick it up without process restart.

get_last_discovery_result() -> DiscoveryResult | None (v0.6.0)

Return the most recent DiscoveryResult produced by discover_plugins() or register_plugins(), or None if neither has been called yet.

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.

deactivate_all() -> None

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

reload_plugin(name: str) -> bool

Hot-reload a single plugin: deactivate, re-import module from disk, re-init, re-activate. Returns True on success. Distinct from rediscover(), which re-reads entry points but does not reload module code for existing plugins.

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 as a {name: message} dict. Since v0.6.0 this is sourced internally from DiscoveryResult.errors. New code should prefer DiscoveryResult directly (returned from discover_plugins / register_plugins, or queried via get_last_discovery_result) for richer per-plugin diagnostics including structured filter_reason values, exception causes, and severity.

Health Checks

health_check() -> dict[str, dict]

Run health() on all active plugins. Exceptions are caught and reported as {"status": "error", "error": "..."}.

Hooks

register_hookspecs(spec_module: object) -> None

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

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

Call a named hook on all registered plugins. Returns a list of results. If the hook is unknown, logs a warning and returns []. If any implementation raises, returns [].

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

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

Introspection

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

Hook names implemented by a specific plugin.

get_all_hook_names() -> list[str]

All registered hook spec names.

inspect_plugin(name: str) -> PluginInspection | None (v0.9.0)

Aggregate state, config, health, hooks, routes, and identity for a plugin into a single frozen PluginInspection snapshot. Single-call replacement for the union of get_plugin / get_last_discovery_result().states[name] / get_plugin_hooks(name) / plugin.health() / plugin.config. Returns None if the plugin name is not known to this manager (never initialized, or purged by rediscover removal / reload_plugin).

Plugin-instance fields (version, config, health, routes, hooks) are populated when an instance is reachable via the lifecycle. For plugins filtered before initialization (e.g. wrong_application, missing_target_application, not_enabled), those fields carry safe defaults; the state field always reflects the real filter outcome.

inspection = pm.inspect_plugin("export")
if inspection is None:
    return  # unknown plugin
print(f"Active: {inspection.state.activated}, activated_at: {inspection.state.activated_at}")
print(f"Source: {inspection.state.source}, hooks: {inspection.hooks}")
print(f"Health: {inspection.health}")

Event Hooks (v0.9.0)

Registration-based callbacks for observing lifecycle transitions. Each registration returns an idempotent deregistration closure: calling it removes the callback. Callbacks fire after the lifecycle step completes successfully; failures inside a callback are logged at WARNING and swallowed so they cannot break the lifecycle pipeline. The subscriber list is snapshotted at dispatch start, so callbacks registered during another callback's execution fire on the next event, not the current dispatch loop.

on_plugin_activated(callback: Callable[[str], None]) -> Callable[[], None]

Register a callback fired after each successful plugin activation. Callback receives the plugin name.

def log_activation(name: str) -> None:
    metrics.increment("plugin.activated", tags={"name": name})

unreg = pm.on_plugin_activated(log_activation)
# Later: unreg() to stop receiving events.

on_plugin_deactivated(callback: Callable[[str], None]) -> Callable[[], None]

Register a callback fired after each successful deactivation (including LIFO traversal during deactivate_all). Callback receives the plugin name.

on_config_refreshed(callback: Callable[[str, dict, dict], None]) -> Callable[[], None]

Register a callback fired per plugin whose on_config_changed completed without raising during refresh_config. Callback receives (plugin_name, old_app_config, new_app_config).

def log_config_refresh(name: str, old: dict, new: dict) -> None:
    print(f"{name} reacted to config change")

pm.on_config_refreshed(log_config_refresh)

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.

FastAPI Integration

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

Mount routes from all active plugins onto a FastAPI application. Idempotent since v0.8.0: calling with the same plugin set on the same app is a no-op. Per-app tracking lives in pluginforge.fastapi_ext._mounted_plugins (WeakKeyDictionary[FastAPI, set[str]]); use pluginforge.testing.IsolatedPluginManager in tests to clear the tracking between sweeps. See FastAPI Integration.

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.

Alembic Integration

collect_migrations() -> dict[str, str]

Collect Alembic migration directories from all active plugins.

Related Types

The v0.6.0 lifecycle work introduced structured types that surface in the return values and errors of the methods above. v0.7.0 added identity-related values; v0.9.0 added PluginInspection.

  • DiscoveryResult — return of discover_plugins() / register_plugins(). Fields: states: dict[str, PluginState], activated: list[str], errors: list[PluginError]. Helpers: by_filter_reason(reason), filtered_out().
  • DiscoveryDiff — return of rediscover(). Fields: added, removed, unchanged (all list[str]), states: dict[str, PluginState], errors: list[PluginError]. Helper (v0.10.0): by_filter_reason(reason) parity with DiscoveryResult.
  • PluginState — per-plugin discovery state. Fields include name, discovered, enabled_in_config, disabled_in_config, activated, filter_reason, load_error, plus v0.9.0 activated_at, last_config_change, source.
  • PluginError — structured error. Fields: name, phase, cause, user_facing_message, severity.
  • PluginInspection (v0.9.0) — return of inspect_plugin(name). Frozen aggregator. Fields: name, version, target_application, min_app_version, state (full PluginState), config, health, hooks, routes, dependencies.
  • FilterReason — Literal of ten values explaining why a plugin was not activated (v0.9.0 added missing_target_application).
  • ErrorPhase — Literal of seven values identifying which lifecycle phase emitted an error.

For the full taxonomy and semantics, see:

Complete Usage

from pluginforge import PluginManager

# Setup with v0.7.0 features
pm = PluginManager(
    config_path="config/app.yaml",
    api_version="1",
    app_version="1.5.0",                  # v0.6.0: enables min_app_version gating
    app_id="my-app",                      # v0.7.0: enables target_application gating
    api_version_severity="warning",       # v0.6.0: how api_version mismatch is handled
    app_version_severity="error",         # v0.6.0: how min_app_version mismatch is handled
)
pm.register_hookspecs(my_hooks)

# Discover and inspect
result = pm.discover_plugins()
print(f"Activated: {result.activated}")
for name, reason in result.filtered_out().items():
    print(f"  {name}: filtered ({reason})")
for err in result.errors:
    print(f"  {err.severity.upper()} {err.name} [{err.phase}]: {err.user_facing_message}")

# Pick up plugins installed mid-process (v0.6.0)
diff = pm.rediscover()
if diff.added:
    print(f"Newly activated: {diff.added}")

# Live-reload the app config and notify plugins (v0.6.0)
refresh_errors = pm.refresh_config()  # reload from disk, fire on_config_changed
for err in refresh_errors:
    print(f"  {err.name}.on_config_changed raised: {err.user_facing_message}")

# Hot-reload a single plugin's module (development workflow)
pm.reload_plugin("export")

# Hooks
results = pm.call_hook("on_startup")

# Extensions
exporters = pm.get_extensions(ExportFormat)

# Health
status = pm.health_check()

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

# Shutdown
pm.deactivate_all()

Clone this wiki locally