Entry Points y sistemas extensibles con plugins en Python

El mecanismo de entry points es el contrato estándar entre un paquete instalado y el ecosistema de paquetes que quieren extenderlo. En vez de que tu aplicación sepa de antemano quién la extiende, los plugins se anuncian al instalarse, y tu aplicación los descubre en runtime consultando los metadatos del entorno.

La idea central: cuando instalas un paquete Python, su pyproject.toml (o el antiguo setup.cfg) le dice al installer que registre ciertos puntos de entrada bajo un nombre de grupo. Ese registro queda en los metadatos del entorno virtual —en el directorio .dist-info del paquete— y importlib.metadata sabe cómo leerlos. Tu aplicación nunca importa el plugin directamente; le pregunta al entorno “¿quién se registró bajo este grupo?” y trabaja con lo que encuentre.

Esto resuelve el problema de extensibilidad sin inversión de dependencias forzada. Si tu app importara plugins directamente, el núcleo dependería de cada extensión, que es exactamente lo opuesto a lo que quieres. Con entry points, la dependencia va en una sola dirección: los plugins declaran que extienden tu app, no al revés. Pytest funciona exactamente así —pytest11 es el grupo bajo el que cualquier paquete puede registrar fixtures, hooks y plugins sin que pytest sepa nada del paquete concreto hasta que lo descubre.

El cuándo es claro: sistemas donde terceros (o tus propios sub-paquetes) necesitan inyectar comportamiento sin modificar el core. Formateadores, validadores, backends de almacenamiento, extractores, comandos CLI adicionales. Si te encuentras manteniendo una lista de strings con nombres de módulos a importar, o un diccionario que crece con cada extensión, entry points es lo que deberías estar usando.

Lo que rompe si lo haces mal: si usas importlib.import_module con rutas hardcodeadas, cualquier plugin nuevo requiere un cambio en el core. Si mantienes un registro global mutable, el orden de importación pasa a ser relevante y los tests se vuelven frágiles. Y si no usas entry points sino simplemente convenciones de nombres (mi_app_plugin_*), pierdes la capacidad de que el instalador gestione el ciclo de vida de los plugins.

# ── mi_app/core/registry.py ──────────────────────────────────────────────
"""
Núcleo de la aplicación: descubre y gestiona plugins de transformación.
No importa ningún plugin concreto. No sabe cuántos hay.
"""
from __future__ import annotations

import logging
from importlib.metadata import entry_points, EntryPoint
from typing import Protocol, runtime_checkable

logger = logging.getLogger(__name__)

GROUP = "mi_app.transformers"


@runtime_checkable
class Transformer(Protocol):
    """Contrato que todo plugin debe satisfacer."""

    name: str  # identificador único legible

    def transform(self, data: str) -> str:
        ...


class PluginRegistry:
    def __init__(self) -> None:
        self._plugins: dict[str, Transformer] = {}

    def load_plugins(self) -> None:
        """
        Descubre todos los plugins instalados bajo el grupo GROUP.
        Cada entry point es una referencia a un objeto importable;
        .load() resuelve la importación en este momento, no antes.
        """
        eps: list[EntryPoint] = entry_points(group=GROUP)

        for ep in eps:
            try:
                plugin_class = ep.load()  # importación diferida — falla aquí, no al startup
                instance = plugin_class()

                if not isinstance(instance, Transformer):
                    # Usamos runtime_checkable para verificar el protocolo en carga
                    logger.warning(
                        "Plugin '%s' no implementa el protocolo Transformer, ignorado.",
                        ep.name,
                    )
                    continue

                self._plugins[ep.name] = instance
                logger.debug("Plugin cargado: %s → %s", ep.name, plugin_class)

            except Exception as exc:  # noqa: BLE001
                # Un plugin roto no debe tumbar la app principal
                logger.error("Error cargando plugin '%s': %s", ep.name, exc)

    def get(self, name: str) -> Transformer | None:
        return self._plugins.get(name)

    def all(self) -> list[Transformer]:
        return list(self._plugins.values())


# ── mi_app/main.py ────────────────────────────────────────────────────────
def run_pipeline(data: str, registry: PluginRegistry) -> dict[str, str]:
    """Aplica todos los transformers registrados al mismo dato."""
    return {t.name: t.transform(data) for t in registry.all()}
# ── pyproject.toml del PLUGIN (paquete separado: mi_app_upper) ────────────
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "mi-app-upper"
version = "0.1.0"
dependencies = []          # no declara dependencia en mi_app; la dirección va al revés

[project.entry-points."mi_app.transformers"]
# clave = "módulo:atributo"
upper = "mi_app_upper.plugin:UpperTransformer"
# ── mi_app_upper/plugin.py ────────────────────────────────────────────────
class UpperTransformer:
    name = "upper"

    def transform(self, data: str) -> str:
        return data.upper()
# ── tests/test_registry.py  (sin instalar nada, usando importlib.metadata stubs) ──
"""
Para tests unitarios del core puedes mockear entry_points directamente.
"""
from unittest.mock import patch, MagicMock
from mi_app.core.registry import PluginRegistry


def _make_ep(name: str, target):
    ep = MagicMock()
    ep.name = name
    ep.load.return_value = target
    return ep


def test_load_plugins_registers_valid_transformer():
    class FakeTransformer:
        name = "fake"
        def transform(self, data: str) -> str:
            return data[::-1]

    fake_ep = _make_ep("reverse", FakeTransformer)

    with patch("mi_app.core.registry.entry_points", return_value=[fake_ep]):
        registry = PluginRegistry()
        registry.load_plugins()

    assert registry.get("reverse") is not None
    assert registry.get("reverse").transform("abc") == "cba"


def test_broken_plugin_does_not_crash_registry():
    bad_ep = MagicMock()
    bad_ep.name = "broken"
    bad_ep.load.side_effect = ImportError("módulo inexistente")

    with patch("mi_app.core.registry.entry_points", return_value=[bad_ep]):
        registry = PluginRegistry()
        registry.load_plugins()  # no debe lanzar excepción

    assert registry.all() == []

Qué significa cada decisión

La separación de paquetes es la arquitectura. El plugin vive en su propio repositorio y su propio ciclo de release. mi_app nunca aparece como dependencia en el pyproject.toml del plugin —solo en el grupo de entry points, que es metadata, no código. Cuando haces pip install mi-app-upper, el installer escribe la entrada en .dist-info/entry_points.txt del paquete, y importlib.metadata la encuentra sin que nadie haya importado nada todavía.

ep.load() es importación diferida deliberada. El entry point es una referencia "módulo:atributo" que no se resuelve hasta que llamas a .load(). Esto te da la oportunidad de envolver la carga en try/except, loguear errores por plugin, y continuar. Si un plugin tiene una dependencia faltante o un error de sintaxis, el resto del sistema sigue funcionando. Pytest hace lo mismo: un plugin roto no cancela toda la sesión de tests.

runtime_checkable + Protocol como contrato explícito. En vez de herencia (que acoplaría los plugins a una clase base de tu paquete), defines un Protocol estructural. El plugin solo necesita tener los atributos y métodos correctos; no importa Transformer para nada. isinstance(instance, Transformer) funciona gracias a @runtime_checkable y te da una segunda línea de defensa si ep.load() devuelve algo que no es lo que esperas.

El grupo es un namespace tuyo. La convención es nombre_paquete.categoria. Si publicas una librería llamada acme, tu grupo podría ser acme.backends o acme.hooks. Nadie más puede registrar en ese grupo sin saberlo —o más bien, si lo hacen, es porque explícitamente quieren extender tu sistema. Pytest usa pytest11; Babel usa babel.extractors; setuptools expone distutils.commands. Son todos el mismo mecanismo.

El test mockea entry_points, no el filesystem. No necesitas crear paquetes instalados para testear el registry. Mockeas la función que devuelve los entry points y pruebas el comportamiento del core en aislamiento. Es la misma técnica que usarías para mockear open o requests.get.

Errores que debes conocer

Error: Registrar el entry point con el nombre del grupo incorrecto hace que el plugin se instale sin errores pero nunca aparezca en los resultados de entry_points(group=...).

# ❌ Wrong — typo en el nombre del grupo
[project.entry-points."mi_app.transformer"]   # falta la 's'
upper = "mi_app_upper.plugin:UpperTransformer"

# ✅ Right
[project.entry-points."mi_app.transformers"]
upper = "mi_app_upper.plugin:UpperTransformer"

entry_points no lanza error cuando un grupo no existe, simplemente devuelve lista vacía. Comprueba el grupo exacto con python -c "from importlib.metadata import entry_points; print(entry_points(group='mi_app.transformers'))" después de instalar.


Error: Llamar a entry_points() sin el argumento group y luego iterar sobre el resultado como si fuera una lista plana — la API cambió en Python 3.12 y el objeto retornado es un SelectableGroups que no se comporta como un dict en versiones anteriores.

# ❌ Wrong — comportamiento inconsistente entre 3.9 y 3.12
all_eps = entry_points()
plugins = all_eps.get("mi_app.transformers", [])   # falla en >=3.12

# ✅ Right — usa siempre el keyword argument group=
plugins = entry_points(group="mi_app.transformers")

La forma con group= existe desde Python 3.9 (vía importlib_metadata backport) y es la única API estable hacia adelante.


Error: Asumir que ep.name coincide con el nombre del paquete. La clave en [project.entry-points] es el nombre del entry point específico, no el nombre del paquete que lo declara.

# ❌ Wrong — confundir ep.name con ep.dist.name
for ep in entry_points(group="mi_app.transformers"):
    print(f"Paquete: {ep.name}")   # imprime "upper", no "mi-app-upper"

# ✅ Right — si necesitas el paquete origen, usa ep.dist
for ep in entry_points(group="mi_app.transformers"):
    package_name = ep.dist.name if ep.dist else "desconocido"
    print(f"Entry point '{ep.name}' provisto por '{package_name}'")

ep.dist da acceso a los metadatos completos del paquete distribuidor, útil para mostrar versiones o para rechazar plugins de versiones incompatibles.

Dejar un comentario

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Scroll al inicio