# Copyright (C) 2023-2026 Pain001. All rights reserved.
# SPDX-License-Identifier: Apache-2.0 OR MIT
#
# Licensed under either of the Apache License, Version 2.0 or the MIT
# License, at your option. You may not use this file except in
# compliance with one of those licences. Copies are provided in
# LICENSE-APACHE and LICENSE-MIT.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the Licences is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the applicable Licence for the specific language
# governing permissions and limitations.
"""Process-level plugin registry + entry-point discovery.
Holds one :class:`PluginRegistry` instance per Python process. The
registry is populated lazily on first use:
1. Built-in plugins shipped inside pain001 register first.
2. External plugins discovered via ``importlib.metadata.entry_points``
register on top, winning over built-ins with the same name.
3. The ``PAIN001_DISABLE_PLUGINS`` env var (comma-separated names)
skips matching plugins regardless of source - the production
escape hatch when a plugin breaks.
Plugin authors talk to the registry only through
:func:`pain001.plugins.registry.registry` (the module-level
singleton); inside pain001, the loader / validator / scheme
dispatchers also go through the same singleton so external plugins
and built-ins share one code path.
"""
from __future__ import annotations
import hashlib
import logging
import os
import traceback
from collections.abc import Iterable
from importlib import metadata
from typing import Any, cast
from pain001.plugins._version import PAIN001_API_VERSION
from pain001.plugins.contracts import (
AbstractLoader,
AbstractScheme,
AbstractValidator,
AbstractWriter,
PluginInfo,
PluginMeta,
)
logger = logging.getLogger(__name__)
ENTRY_POINT_GROUPS: dict[str, str] = {
"loader": "pain001.loaders",
"validator": "pain001.validators",
"scheme": "pain001.schemes",
"writer": "pain001.writers",
}
[docs]
class PluginRegistryError(RuntimeError):
"""Raised when the registry cannot honour a lookup or registration."""
[docs]
class PluginVersionError(PluginRegistryError):
"""Raised when a plugin declares an API version newer than this build."""
[docs]
class PluginRegistry:
"""In-process registry of every loaded plugin, keyed by kind + name.
Not thread-safe for *registration* (plugins are registered once
at startup and the registry is read-only after); fully safe for
concurrent lookups thereafter.
Args:
api_version: The ``(major, minor)`` tuple the host pain001
build implements. Plugins declaring a higher *major* are
refused; plugins declaring a higher *minor* are loaded
with a warning.
"""
def __init__(self, api_version: tuple[int, int]) -> None:
self._api_version = api_version
self._loaders: dict[str, AbstractLoader] = {}
self._validators: dict[str, AbstractValidator] = {}
self._schemes: dict[str, AbstractScheme] = {}
self._writers: dict[str, AbstractWriter] = {}
self._disabled = _parse_disabled_env()
self._populated = False
# ---------------------------------------------------------------
# Public lookup API
# ---------------------------------------------------------------
[docs]
def get_loader_for_extension(
self, extension: str
) -> AbstractLoader | None:
"""Return the loader registered for ``extension`` (with leading dot).
Args:
extension: File extension to dispatch on, with the leading
dot, case-insensitive (``".xlsx"``, ``".XLSX"`` both
work).
Returns:
The matching :class:`AbstractLoader`, or ``None`` if none
is registered.
"""
self._ensure_populated()
ext = extension.lower()
for loader in self._loaders.values():
if ext in loader.extensions:
return loader
return None
[docs]
def get_loader(self, name: str) -> AbstractLoader | None:
"""Return the loader registered under ``name`` (or ``None``)."""
self._ensure_populated()
return self._loaders.get(name)
[docs]
def get_validator(self, name: str) -> AbstractValidator | None:
"""Return the validator registered under ``name`` (or ``None``)."""
self._ensure_populated()
return self._validators.get(name)
[docs]
def get_scheme(self, name: str) -> AbstractScheme | None:
"""Return the scheme registered under ``name`` (or ``None``)."""
self._ensure_populated()
return self._schemes.get(name)
[docs]
def get_writer(self, name: str) -> AbstractWriter | None:
"""Return the writer registered under ``name`` (or ``None``)."""
self._ensure_populated()
return self._writers.get(name)
[docs]
def list_plugins(self, kind: str | None = None) -> list[PluginInfo]:
"""Return a flat list of every registered plugin's metadata.
Args:
kind: Optional filter (``"loader"``, ``"validator"``,
``"scheme"``, ``"writer"``). When ``None``, every
kind is included.
Returns:
A list of :class:`PluginInfo`, sorted by ``(kind, name)``.
"""
self._ensure_populated()
infos: list[PluginInfo] = []
for k, store in self._stores().items():
if kind is not None and k != kind:
continue
for plugin in store.values():
infos.append(PluginInfo(kind=k, meta=plugin.meta))
infos.sort(key=lambda info: (info.kind, info.meta.name))
return infos
# ---------------------------------------------------------------
# Registration API (called by built-ins and entry-point loader)
# ---------------------------------------------------------------
[docs]
def register_loader(self, loader: AbstractLoader) -> None:
"""Register ``loader`` under its declared name; later wins."""
self._register(loader, self._loaders, kind="loader")
[docs]
def register_validator(self, validator: AbstractValidator) -> None:
"""Register ``validator`` under its declared name; later wins."""
self._register(validator, self._validators, kind="validator")
[docs]
def register_scheme(self, scheme: AbstractScheme) -> None:
"""Register ``scheme`` under its declared name; later wins."""
self._register(scheme, self._schemes, kind="scheme")
[docs]
def register_writer(self, writer: AbstractWriter) -> None:
"""Register ``writer`` under its declared name; later wins."""
self._register(writer, self._writers, kind="writer")
[docs]
def reset(self) -> None:
"""Drop every plugin and mark the registry unpopulated again.
Test-only seam. Production code never resets the singleton.
"""
self._loaders.clear()
self._validators.clear()
self._schemes.clear()
self._writers.clear()
self._disabled = _parse_disabled_env()
self._populated = False
# ---------------------------------------------------------------
# Internals
# ---------------------------------------------------------------
def _stores(self) -> dict[str, dict[str, Any]]:
"""Return the four plugin stores keyed by kind name."""
return {
"loader": self._loaders,
"validator": self._validators,
"scheme": self._schemes,
"writer": self._writers,
}
def _register(
self,
plugin: Any,
store: dict[str, Any],
*,
kind: str,
) -> None:
"""Validate ``plugin`` against the contract and stash it in ``store``."""
meta = getattr(plugin, "meta", None)
if not isinstance(meta, PluginMeta):
raise PluginRegistryError(
f"{kind} plugin {plugin!r} missing a PluginMeta `meta` "
"attribute"
)
if meta.name in self._disabled:
logger.info(
"skipping disabled %s plugin %s "
"(set via PAIN001_DISABLE_PLUGINS)",
kind,
meta.name,
)
return
self._check_api_compatibility(meta, kind=kind)
if meta.name in store and store[meta.name] is not plugin:
logger.warning(
"%s plugin %s from %s overrides existing %s",
kind,
meta.name,
meta.source,
store[meta.name].meta.source,
)
store[meta.name] = plugin
def _check_api_compatibility(self, meta: PluginMeta, *, kind: str) -> None:
"""Refuse plugins ahead of the host's major; warn on minor gap."""
host_major, host_minor = self._api_version
plugin_major, plugin_minor = meta.api_version
if plugin_major > host_major:
raise PluginVersionError(
f"{kind} plugin {meta.name!r} targets API "
f"v{plugin_major}.{plugin_minor}, but this pain001 only "
f"supports up to v{host_major}.{host_minor}; upgrade pain001"
)
if (plugin_major, plugin_minor) > (host_major, host_minor):
logger.warning(
"%s plugin %s targets API v%d.%d; this pain001 implements "
"v%d.%d - newer plugin methods may be ignored",
kind,
meta.name,
plugin_major,
plugin_minor,
host_major,
host_minor,
)
def _ensure_populated(self) -> None:
"""Lazily register built-ins + entry-point plugins on first lookup."""
if self._populated:
return
self._populated = True # set before, so re-entrant lookups don't loop
# Built-ins first; entry-point plugins later (later wins).
from pain001.plugins import _builtins # noqa: PLC0415 - lazy
_builtins.register_all(self)
_load_entry_point_plugins(self)
def _parse_disabled_env() -> frozenset[str]:
"""Read ``PAIN001_DISABLE_PLUGINS`` into a set of names (or empty)."""
raw = os.environ.get("PAIN001_DISABLE_PLUGINS", "")
return frozenset(s.strip() for s in raw.split(",") if s.strip())
def _load_entry_point_plugins(reg: PluginRegistry) -> None:
"""Walk each entry-point group and register what we find.
A plugin that raises during instantiation is skipped with a
warning so one broken plugin can't take pain001 down. The
operator can isolate the culprit via ``pain001 plugins list``
plus the logged plugin name.
"""
for kind, group in ENTRY_POINT_GROUPS.items():
for entry in _iter_entry_points(group):
try:
plugin_cls = entry.load()
plugin = plugin_cls()
# Stamp the source so list_plugins shows the dist name.
_stamp_source(plugin, entry)
_register_by_kind(reg, plugin, kind)
except Exception as exc:
# Exception messages can contain payment data or credentials.
# Identify the failure by type and stack hash without logging
# the message, local variables, or the raw traceback.
exception_class = type(exc).__name__
traceback_hash = hashlib.sha256(
"".join(traceback.format_tb(exc.__traceback__)).encode()
).hexdigest()
logger.warning(
"skipping %s plugin from entry-point %s: %s "
"(traceback_sha256=%s)",
kind,
entry.name,
exception_class,
traceback_hash,
extra={
"plugin_name": entry.name,
"plugin_kind": kind,
"exception_class": exception_class,
"traceback_hash": traceback_hash,
},
)
def _iter_entry_points(group: str) -> Iterable[metadata.EntryPoint]:
"""Yield entry points for ``group``; tolerate older metadata APIs.
Python 3.10+ exposes ``EntryPoints.select(...)``; older APIs return
a ``dict[str, list[EntryPoint]]``. We support both so a plugin
rollout can't be blocked by an upstream `importlib.metadata`
shape change.
"""
eps = metadata.entry_points()
select = getattr(eps, "select", None)
if callable(select):
return cast("Iterable[metadata.EntryPoint]", select(group=group))
# Legacy importlib.metadata exposed a ``dict`` mapping group -> list.
return cast( # pragma: no cover - legacy
"dict[str, Iterable[metadata.EntryPoint]]", eps
).get(group, [])
def _stamp_source(plugin: Any, entry: metadata.EntryPoint) -> None:
"""Replace ``meta.source`` with the entry point's distribution name.
Plugin authors must not set ``source`` themselves; this is the
one piece of metadata the registry owns so an operator can trust
it when auditing what's loaded.
"""
dist = getattr(entry, "dist", None)
dist_name = getattr(dist, "name", entry.name) or entry.name
dist_version = getattr(dist, "version", "?") or "?"
source = f"{dist_name}=={dist_version}"
# PluginMeta is frozen; rebuild it with the stamped source.
new_meta = PluginMeta(
name=plugin.meta.name,
version=plugin.meta.version,
description=plugin.meta.description,
api_version=plugin.meta.api_version,
source=source,
)
object.__setattr__(plugin, "meta", new_meta)
def _register_by_kind(reg: PluginRegistry, plugin: Any, kind: str) -> None:
"""Dispatch to the right ``register_*`` method based on plugin kind."""
if kind == "loader":
reg.register_loader(plugin)
elif kind == "validator":
reg.register_validator(plugin)
elif kind == "scheme":
reg.register_scheme(plugin)
elif kind == "writer":
reg.register_writer(plugin)
else: # pragma: no cover - defensive
raise PluginRegistryError(f"unknown plugin kind: {kind!r}")
registry: PluginRegistry = PluginRegistry(PAIN001_API_VERSION)
"""Process-level singleton; everyone reads through this one."""