"""Base plugin abstract class with lifecycle hooks.""" from __future__ import annotations from abc import ABC from typing import TYPE_CHECKING, Any from fastapi import APIRouter from app.plugins.manifest import PluginManifest if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession from app.core.event_bus import EventBus from app.core.service_container import ServiceContainer class BasePlugin(ABC): """Abstract base class for all LeoCRM plugins. Subclasses must define a ``manifest`` class attribute (PluginManifest) and implement the lifecycle hooks. """ manifest: PluginManifest def __init__(self) -> None: if not hasattr(self, "manifest") or self.manifest is None: raise ValueError(f"{self.__class__.__name__} must define a 'manifest' attribute") self._event_handlers: dict[str, Any] = {} self._routers: list[APIRouter] = [] self._container: Any = None # ServiceContainer, set during on_activate @property def services(self) -> Any: """Access the ServiceContainer after activation.""" if self._container is None: raise RuntimeError("Services not available — plugin not activated") return self._container # ─── Lifecycle Hooks ─── async def on_install(self, db: AsyncSession, service_container: ServiceContainer) -> None: """Called when the plugin is installed (after migrations are run). Override to perform seed data or initial setup. """ return None async def on_activate( self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus ) -> None: """Called when the plugin is activated. Default implementation subscribes to events listed in the manifest and registers manifest MiniApps (Phase M1): ``miniapps`` contributions plus ``dashboard_widgets`` entries (alias — one contribution type, #359 philosophy). Registered automatically here; no per-plugin code needed. """ for event_name in self.manifest.events: handler = self._make_event_handler(event_name) self._event_handlers[event_name] = handler event_bus.subscribe(event_name, handler) self._container = service_container self._register_manifest_miniapps() async def on_deactivate( self, db: AsyncSession, service_container: ServiceContainer, event_bus: EventBus ) -> None: """Called when the plugin is deactivated. Override to clean up runtime state. Default implementation unsubscribes all event listeners and hooks that were registered during activation. """ # Unsubscribe event listeners for event_name, handler in self._event_handlers.items(): event_bus.unsubscribe(event_name, handler) self._event_handlers.clear() # Unregister all hooks owned by this plugin from app.core.hooks import get_hook_registry get_hook_registry().unregister_all_for_plugin(self.manifest.name) # Unregister MiniApps owned by this plugin (Phase M1) from app.plugins.miniapp_registry import get_miniapp_registry get_miniapp_registry().unregister_plugin(self.manifest.name) def _register_manifest_miniapps(self) -> None: """Register manifest MiniApps in the universal registry (Phase M1). Sources: - ``manifest.miniapps`` — native MiniApp contributions - ``manifest.dashboard_widgets`` — alias: FrontendDashboardWidget entries become MiniApps with component path + spans + permission so existing plugin manifests keep working without changes. """ from app.plugins.miniapp_registry import get_miniapp_registry registry = get_miniapp_registry() name = self.manifest.name for m in getattr(self.manifest, "miniapps", None) or []: registry.register( app_id=m.app_id, name=m.name, icon=m.icon, description=m.description, plugin_name=name, render_schema=m.render_schema, permission=getattr(m, "permission", ""), settings_schema=getattr(m, "settings_schema", {}), col_span=getattr(m, "col_span", 1), row_span=getattr(m, "row_span", 1), hosts=getattr(m, "hosts", None), component=getattr(m, "component", ""), order=getattr(m, "order", 100), ) for w in getattr(self.manifest, "dashboard_widgets", None) or []: registry.register( app_id=w.id, name=w.label or w.id, icon=w.icon, description="", plugin_name=name, permission=w.permission, col_span=w.col_span, row_span=w.row_span, hosts=["chat", "dashboard", "window"], component=w.component, order=w.order, ) async def on_uninstall(self, db: AsyncSession, service_container: ServiceContainer) -> None: """Called when the plugin is uninstalled (before data tables are dropped). Override to clean up any external resources. """ return None # ─── Notification Types ─── def get_notification_types(self) -> list[dict[str, Any]]: """Return notification types this plugin registers. Override in subclasses to register notification types. Each dict must have: type_key, category, label, description, is_enabled_by_default """ return [] async def register_event_handlers(self, event_bus: EventBus) -> None: """Register event handlers for the background worker (ARCH-038 hook). The worker calls this on every active plugin at startup so plugins can subscribe to events even when the web process is separate. Default: no-op. Override to subscribe handlers. """ return None # ─── Job Modules ─── def get_job_modules(self) -> list[str]: """Return list of ARQ job module paths to import for background workers. Override in subclasses that register background jobs. The worker imports each module so that register_job() calls fire. Default: no job modules. """ return [] # ─── Entity Models ─── def get_entity_models(self) -> dict[str, type]: """Return entity_type → SQLAlchemy model class mapping for permission system. Override in subclasses that own entities with OwnedMixin. These models are registered in ENTITY_MODELS at activation time so the permission system can resolve entity types dynamically. Default: no entity models. """ return {} # ─── Field Definitions ─── def get_field_definitions(self) -> list[dict[str, str]]: """Return field definitions from manifest.""" return [ {"module": fd.module, "field": fd.field, "label": fd.label, "sensitivity": fd.sensitivity} for fd in self.manifest.field_definitions ] # ─── Route Registration ─── def get_routes(self) -> list[APIRouter]: """Return list of APIRouter instances to mount on the FastAPI app. Default implementation loads routers from manifest route definitions by importing the module and getting the specified attribute. """ if self._routers: return self._routers routers: list[APIRouter] = [] for route_def in self.manifest.routes: import importlib module = importlib.import_module(route_def.module) router: APIRouter = getattr(module, route_def.router_attr) routers.append(router) self._routers = routers return routers # ─── Event Handler Factory ─── def _make_event_handler(self, event_name: str) -> Any: """Create an event handler coroutine for the given event name. Looks for a method named ``on_`` with dots replaced by underscores. Falls back to ``on_event`` if it exists, otherwise uses a no-op handler. """ method_name = f"on_{event_name.replace('.', '_')}" specific = getattr(self, method_name, None) if specific is not None: return specific fallback = getattr(self, "on_event", None) if fallback is not None: return fallback # Default no-op handler async def _noop_handler(payload: dict[str, Any]) -> None: pass return _noop_handler # ─── Utility ─── @property def name(self) -> str: return self.manifest.name @property def version(self) -> str: return self.manifest.version def __repr__(self) -> str: return ( f"<{self.__class__.__name__} name={self.manifest.name} version={self.manifest.version}>" )