Examples

Examples

Minimal Plugin

from pluginforge import BasePlugin

class MinimalPlugin(BasePlugin):
    name = "minimal"

This is a valid plugin. All lifecycle methods have default implementations.

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/plugins/health/status

Full Application

# app.py
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: ...

# Setup
app = FastAPI()
pm = PluginManager("config/app.yaml")
pm.register_hookspecs(MyHookSpec)
pm.discover_plugins()
pm.mount_routes(app)

@app.on_event("startup")
async def startup():
    pm.call_hook("on_startup")

@app.on_event("shutdown")
async def shutdown():
    pm.call_hook("on_shutdown")
    pm.deactivate_all()

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

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