Examples

Examples

Minimal Plugin

from pluginforge import BasePlugin

class MinimalPlugin(BasePlugin):
    name = "minimal"

This is a valid plugin. All lifecycle methods have default implementations. Since v0.9.0, plugins without target_application are filtered when the host has adopted app_id; see the next example for the recommended shape.

Minimal Plugin (v0.7.0 recommended)

from pluginforge import BasePlugin

class MinimalPlugin(BasePlugin):
    name = "minimal"
    target_application = "myapp"     # v0.7.0: declare which host you target

When the host runs with PluginManager(app_id="myapp"), this plugin activates without identity-related warnings. When the host has not adopted app_id, the plugin still activates (claim unvalidated).

Plugin with Config

from pluginforge import BasePlugin

class NotifierPlugin(BasePlugin):
    name = "notifier"
    version = "1.0.0"
    description = "Send notifications"

    def activate(self) -> None:
        self.channel = self.config.get("channel", "email")
        self.recipients = self.config.get("recipients", [])

# config/plugins/notifier.yaml
channel: "slack"
recipients:
  - "#general"
  - "#alerts"

Plugin with Dependencies

class AnalyticsPlugin(BasePlugin):
    name = "analytics"
    depends_on = ["storage"]

    def activate(self) -> None:
        # "storage" is guaranteed to be active at this point
        pass

Plugin with Hooks

import pluggy
from pluginforge import BasePlugin

hookimpl = pluggy.HookimplMarker("myapp")

class AuditPlugin(BasePlugin):
    name = "audit"

    def activate(self) -> None:
        self.log = []

    @hookimpl
    def on_document_save(self, document: dict) -> None:
        self.log.append({"action": "save", "doc": document["title"]})

    @hookimpl
    def on_document_delete(self, document_id: str) -> None:
        self.log.append({"action": "delete", "id": document_id})

Plugin with FastAPI Routes

from fastapi import APIRouter
from pluginforge import BasePlugin

class HealthPlugin(BasePlugin):
    name = "health"

    def get_routes(self) -> list:
        router = APIRouter()

        @router.get("/status")
        def status():
            return {"status": "ok", "plugin_version": self.version}

        return [router]

Accessible at: GET /api/status (mount_routes prepends /api; the router has no internal prefix, so the route lands at /api/status rather than under a per-plugin namespace. To namespace by plugin, give the router an explicit prefix="/health" — that yields /api/health/status.)

Full Application

# app.py
from contextlib import asynccontextmanager

import pluggy
from fastapi import FastAPI
from pluginforge import PluginManager

# Define hook specs
hookspec = pluggy.HookspecMarker("myapp")

class MyHookSpec:
    @hookspec
    def on_startup(self) -> None: ...

    @hookspec
    def on_shutdown(self) -> None: ...

pm = PluginManager(
    "config/app.yaml",
    app_version="1.0.0",   # v0.6.0: enables min_app_version gating
    app_id="myapp",        # v0.7.0: enables target_application gating
)
pm.register_hookspecs(MyHookSpec)

@asynccontextmanager
async def lifespan(app: FastAPI):
    result = pm.discover_plugins()
    pm.mount_routes(app)
    pm.call_hook("on_startup")
    yield
    pm.call_hook("on_shutdown")
    pm.deactivate_all()

app = FastAPI(lifespan=lifespan)

# config/app.yaml
app:
  name: "MyApp"
  version: "1.0.0"
  default_language: "en"

plugins:
  entry_point_group: "myapp.plugins"
  enabled:
    - "health"
    - "audit"