♻️ Plugin system for fastapi app (yes, this time it's actually wired up) 😅

Author: craforge

src/fastapi_cli/cli.py

@@ -1,11 +1,13 @@
 import logging
+from importlib.metadata import entry_points as _entry_points
 from pathlib import Path
 from typing import Annotated, Any
 
 import typer
 from pydantic import ValidationError
 from rich import print
 from rich.tree import Tree
+from typer.models import CommandInfo
 
 from fastapi_cli.config import FastAPIConfig
 from fastapi_cli.discover import get_import_data, get_import_data_from_import_string
@@ -15,9 +17,7 @@
 from .logging import setup_logging
 from .utils.cli import get_rich_toolkit, get_uvicorn_log_config
 
-app = typer.Typer(
-    rich_markup_mode="rich", context_settings={"help_option_names": ["-h", "--help"]}
-)
+app = typer.Typer(rich_markup_mode="rich", context_settings={"help_option_names": ["-h", "--help"]})
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +48,55 @@
     pass
 
 
+def _cmd_name(registered_command: CommandInfo) -> Any:
+    """Return the effective CLI name for a registered Typer command."""
+    if registered_command.name is not None:
+        return registered_command.name
+    if registered_command.callback is not None:
+        return registered_command.callback.__name__.lower().replace("_", "-")
+    return None
+
+
+def _load_cli_plugins(typer_app: typer.Typer) -> None:
+    """Load commands registered via the 'fastapi_cli.plugins' entry point group."""
+
+    # Seed with built-in command names so plugins overriding them get flagged.
+    known: set[str] = set()
+    for registered_command in typer_app.registered_commands:
+        name = _cmd_name(registered_command)
+        if name:
+            known.add(name)
+
+    for entry_point in _entry_points(group="fastapi_cli.plugins"):
+        # Snapshot length to slice off only what the plugin adds.
+        cursor = len(typer_app.registered_commands)
+        try:
+            # resolves the `register` callable.
+            entry_point.load()(typer_app)
+        except Exception as e:
+            # Warning on broken plugin ans continue CLI execution.
+            logger.warning("Plugin '%s' failed to load: %s", entry_point.name, e)
+            continue
+
+        # Walk only plugin's new commands to detect collision.
+        collisions: list[str] = []
+        for registered_command in typer_app.registered_commands[cursor:]:
+            name = _cmd_name(registered_command)
+            if not name:
+                continue
+            if name in known:
+                collisions.append(name)
+            known.add(name)
+
+        # One warning per plugin, listing all the names it overrode.
+        if collisions:
+            logger.warning(
+                "Plugin '%s' overrides existing command(s): %s",
+                entry_point.name,
+                ", ".join(sorted(collisions)),
+            )
+
+
 def version_callback(value: bool) -> None:
     if value:
         print(f"FastAPI CLI version: [green]{__version__}[/green]")
@@ -58,9 +107,7 @@ def version_callback(value: bool) -> None:
 def callback(
     version: Annotated[
         bool | None,
-        typer.Option(
-            "--version", help="Show the version and exit.", callback=version_callback
-        ),
+        typer.Option("--version", help="Show the version and exit.", callback=version_callback),
     ] = None,
     verbose: bool = typer.Option(False, help="Enable verbose output"),
 ) -> None:
@@ -88,9 +135,7 @@ def _get_module_tree(module_paths: list[Path]) -> Tree:
 
     tree = root_tree
     for sub_path in module_paths[1:]:
-        sub_name = (
-            f"🐍 {sub_path.name}" if sub_path.is_file() else f"📁 {sub_path.name}"
-        )
+        sub_name = f"🐍 {sub_path.name}" if sub_path.is_file() else f"📁 {sub_path.name}"
         tree = tree.add(sub_name)
         if sub_path.is_dir():
             tree.add("[dim]🐍 __init__.py[/dim]")
@@ -125,9 +170,7 @@ def _run(
 
         if entrypoint and (path or app):
             toolkit.print_line()
-            toolkit.print(
-                "[error]Cannot use --entrypoint together with path or --app arguments"
-            )
+            toolkit.print("[error]Cannot use --entrypoint together with path or --app arguments")
             toolkit.print_line()
             raise typer.Exit(code=1)
 
@@ -221,9 +264,7 @@ def _run(
             port=port,
             reload=reload,
             reload_dirs=(
-                [str(directory.resolve()) for directory in reload_dirs]
-                if reload_dirs
-                else None
+                [str(directory.resolve()) for directory in reload_dirs] if reload_dirs else None
             ),
             workers=workers,
             root_path=root_path,
@@ -448,4 +489,5 @@ def run(
 
 
 def main() -> None:
+    _load_cli_plugins(app)
     app()

tests/assets/plugins/__init__.py

@@ -0,0 +1,5 @@
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    raise RuntimeError("intentionally broken plugin")

tests/assets/plugins/broken.py

@@ -0,0 +1,7 @@
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    @app.command("dev")  # collides with built-in dev command
+    def dev() -> None:
+        pass  # pragma: no cover

tests/assets/plugins/colliding.py

@@ -0,0 +1,8 @@
+import typer
+
+
+def register(app: typer.Typer) -> None:
+    @app.command("ping")
+    def ping() -> None:
+        """Test command added by plugin."""
+        typer.echo("pong")  # pragma: no cover

tests/assets/plugins/sample.py

@@ -0,0 +1,111 @@
+import sys
+from collections.abc import Generator
+from importlib.metadata import EntryPoint
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+import typer
+
+from fastapi_cli.cli import _load_cli_plugins
+
+assets_path = Path(__file__).parent / "assets"
+
+
+@pytest.fixture
+def plugins_on_path() -> Generator[None, None, None]:
+    original_path = sys.path.copy()
+    sys.path.insert(0, str(assets_path))
+    try:
+        yield
+    finally:
+        sys.path[:] = original_path
+        for key in list(sys.modules.keys()):
+            if key.startswith("plugins."):
+                del sys.modules[key]
+
+
+@pytest.fixture
+def app() -> typer.Typer:
+    app = typer.Typer()
+    return app
+
+
+def _entry_point(name: str, module_attr: str) -> EntryPoint:
+    return EntryPoint(
+        name=name,
+        value=f"plugins.{module_attr}",
+        group="fastapi_cli.plugins",
+    )
+
+
+def test_load_cli_plugins_happy_path(
+    plugins_on_path: None,
+    app: typer.Typer,
+) -> None:
+    entry_point = _entry_point("sample", "sample:register")
+
+    with patch("fastapi_cli.cli._entry_points", return_value=[entry_point]):
+        _load_cli_plugins(app)
+
+    names = {registered_command.name for registered_command in app.registered_commands}
+    assert "ping" in names
+
+
+def test_load_cli_plugins_logs_on_failure(
+    plugins_on_path: None,
+    app: typer.Typer,
+) -> None:
+    entry_point = _entry_point("broken", "broken:register")
+
+    with (
+        patch("fastapi_cli.cli._entry_points", return_value=[entry_point]),
+        patch("fastapi_cli.cli.logger") as mock_logger,
+    ):
+        _load_cli_plugins(app)
+
+    mock_logger.warning.assert_called_once()
+    _fmt, entry_point_name, *_ = mock_logger.warning.call_args.args
+    assert entry_point_name == "broken"
+
+
+def test_load_cli_plugins_warns_on_collision_with_builtin(
+    plugins_on_path: None,
+    app: typer.Typer,
+) -> None:
+
+    @app.command("dev")
+    def existing() -> None:
+        pass  # pragma: no cover
+
+    entry_point = _entry_point("colliding", "colliding:register")
+
+    with (
+        patch("fastapi_cli.cli._entry_points", return_value=[entry_point]),
+        patch("fastapi_cli.cli.logger") as mock_logger,
+    ):
+        _load_cli_plugins(app)
+
+    mock_logger.warning.assert_called_once()
+    _fmt, entry_point_name, collisions = mock_logger.warning.call_args.args
+    assert entry_point_name == "colliding"
+    assert "dev" in collisions
+
+
+def test_load_cli_plugins_warns_on_cross_plugin_collision(
+    plugins_on_path: None,
+    app: typer.Typer,
+) -> None:
+    first = _entry_point("sample", "sample:register")
+    duplicate = _entry_point("duplicate", "sample:register")
+
+    with (
+        patch("fastapi_cli.cli._entry_points", return_value=[first, duplicate]),
+        patch("fastapi_cli.cli.logger") as mock_logger,
+    ):
+        _load_cli_plugins(app)
+
+    mock_logger.warning.assert_called_once()
+    _fmt, entry_point_name, collisions = mock_logger.warning.call_args.args
+    assert entry_point_name == "duplicate"
+    assert "ping" in collisions