Files

282 lines
9.6 KiB
Python
Raw Permalink Normal View History

"""Plugin service layer — business logic for plugin lifecycle operations."""
from __future__ import annotations
import logging
import uuid
from typing import Any
from fastapi import HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.audit import log_audit
from app.plugins.migration_runner import MigrationValidationError
from app.plugins.registry import PluginRegistry, get_registry
logger = logging.getLogger(__name__)
class PluginService:
"""Service layer for plugin lifecycle management.
Wraps the PluginRegistry with audit logging and error handling.
"""
def __init__(self, registry: PluginRegistry | None = None) -> None:
self._registry = registry or get_registry()
@property
def registry(self) -> PluginRegistry:
return self._registry
async def list_plugins(self, db: AsyncSession) -> list[dict[str, Any]]:
"""List all discovered and installed plugins with their status."""
return await self._registry.list_plugins(db)
async def install_plugin(
self,
db: AsyncSession,
name: str,
tenant_id: uuid.UUID | None = None,
user_id: uuid.UUID | None = None,
) -> dict[str, Any]:
"""Install a plugin by name.
Runs migrations, calls on_install hook, creates DB record.
Idempotent: returns existing record if already installed.
Checks that all declared dependencies are installed first.
"""
try:
record = await self._registry.install(db, name)
if tenant_id and user_id:
await log_audit(
db,
tenant_id,
user_id,
action="plugin.install",
entity_type="plugin",
entity_id=getattr(record, "id", None),
changes={"name": name, "version": record.version},
)
return {
"name": record.name,
"display_name": record.display_name,
"version": record.version,
"status": record.status,
"installed": record.installed,
"active": record.active,
"message": "Plugin installed successfully",
}
except ValueError as exc:
msg = str(exc)
if "dependencies" in msg.lower():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=msg,
) from None
raise ValueError(msg) from None
except MigrationValidationError as exc:
raise MigrationValidationError(str(exc)) from None
async def activate_plugin(
self,
db: AsyncSession,
name: str,
tenant_id: uuid.UUID | None = None,
user_id: uuid.UUID | None = None,
) -> dict[str, Any]:
"""Activate a plugin by name.
Registers event listeners and routes. Idempotent.
Checks that all declared dependencies are active first.
"""
try:
record = await self._registry.activate(db, name)
was_already_active = record.active and record.status == "active"
if tenant_id and user_id:
await log_audit(
db,
tenant_id,
user_id,
action="plugin.activate",
entity_type="plugin",
entity_id=getattr(record, "id", None),
changes={"name": name, "was_already_active": was_already_active},
)
return {
"name": record.name,
"display_name": record.display_name,
"version": record.version,
"status": record.status,
"installed": record.installed,
"active": record.active,
"message": "Plugin is already active"
if was_already_active
else "Plugin activated successfully",
}
except ValueError as exc:
msg = str(exc)
if "dependencies" in msg.lower():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=msg,
) from None
raise ValueError(msg) from None
async def deactivate_plugin(
self,
db: AsyncSession,
name: str,
tenant_id: uuid.UUID | None = None,
user_id: uuid.UUID | None = None,
) -> dict[str, Any]:
"""Deactivate a plugin by name.
Unregisters event listeners and routes. Idempotent.
"""
try:
record = await self._registry.deactivate(db, name)
was_already_inactive = not record.active and record.status == "inactive"
if tenant_id and user_id:
await log_audit(
db,
tenant_id,
user_id,
action="plugin.deactivate",
entity_type="plugin",
entity_id=getattr(record, "id", None),
changes={"name": name, "was_already_inactive": was_already_inactive},
)
return {
"name": record.name,
"display_name": record.display_name,
"version": record.version,
"status": record.status,
"installed": record.installed,
"active": record.active,
"message": "Plugin is already inactive"
if was_already_inactive
else "Plugin deactivated successfully",
}
except ValueError as exc:
raise ValueError(str(exc)) from None
async def uninstall_plugin(
self,
db: AsyncSession,
name: str,
remove_data: bool = False,
tenant_id: uuid.UUID | None = None,
user_id: uuid.UUID | None = None,
) -> dict[str, Any]:
"""Uninstall a plugin by name.
Deactivates, calls on_uninstall hook, optionally drops tables, removes DB record.
"""
try:
record = await self._registry.uninstall(db, name, remove_data=remove_data)
dropped_tables = getattr(record, "dropped_tables", [])
if tenant_id and user_id:
await log_audit(
db,
tenant_id,
user_id,
action="plugin.uninstall",
entity_type="plugin",
changes={
"name": name,
"remove_data": remove_data,
"dropped_tables": dropped_tables,
},
)
return {
"name": name,
"display_name": record.display_name,
"version": record.version,
"status": "uninstalled",
"installed": False,
"active": False,
"dropped_tables": dropped_tables,
"message": f"Plugin uninstalled{' and data tables dropped' if remove_data else ''}",
}
except ValueError as exc:
raise ValueError(str(exc)) from None
async def get_plugin_config(self, db: AsyncSession, name: str) -> dict[str, Any]:
"""Get the configuration for a plugin.
Returns the plugin's config field parsed from JSON string.
"""
record = await self._registry._get_plugin_record(db, name)
if record is None:
raise ValueError(f"Plugin '{name}' is not installed")
import json
config_str = getattr(record, "config", None) or "{}"
try:
return json.loads(config_str) if isinstance(config_str, str) else config_str or {}
except (json.JSONDecodeError, TypeError):
return {}
async def update_plugin_config(
self,
db: AsyncSession,
name: str,
config: dict[str, Any],
tenant_id: uuid.UUID | None = None,
user_id: uuid.UUID | None = None,
) -> dict[str, Any]:
"""Update the configuration for a plugin.
Stores the config as a JSON string in the plugin's config field.
"""
import json
record = await self._registry._get_plugin_record(db, name)
if record is None:
raise ValueError(f"Plugin '{name}' is not installed")
record.config = json.dumps(config)
await db.flush()
if tenant_id and user_id:
await log_audit(
db,
tenant_id,
user_id,
action="plugin.update_config",
entity_type="plugin",
entity_id=getattr(record, "id", None),
changes={"name": name, "config": config},
)
return {
"name": name,
"config": config,
"message": "Plugin configuration updated successfully",
}
def get_manifest_schema(self) -> dict[str, Any]:
"""Return the manifest schema documentation."""
from app.plugins.manifest import MANIFEST_SCHEMA_DOC
return MANIFEST_SCHEMA_DOC.model_dump()
# Global service instance
_service: PluginService | None = None
def get_plugin_service() -> PluginService:
"""Get the global plugin service instance."""
global _service
if _service is None:
_service = PluginService()
return _service
def reset_plugin_service_for_testing(registry: PluginRegistry | None = None) -> PluginService:
"""Create a fresh plugin service for testing."""
global _service
_service = PluginService(registry=registry)
return _service