52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""
|
|
BaseAction — interface every action plugin must implement.
|
|
|
|
execute() receives the device, its parent object, params from the request,
|
|
and an emit() callback to write progress messages to the SSE stream.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Awaitable, Callable
|
|
|
|
from app.models.device import Device
|
|
from app.models.object import Object
|
|
|
|
# Type alias for the progress callback
|
|
Emitter = Callable[[str, str], Awaitable[None]] # (message, level)
|
|
|
|
|
|
@dataclass
|
|
class ActionResult:
|
|
success: bool
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
exit_code: int = 0
|
|
data: dict[str, Any] = field(default_factory=dict)
|
|
error: str = ""
|
|
|
|
|
|
class BaseAction(ABC):
|
|
# Unique identifier used in POST /api/v1/tasks {"type": "<name>"}
|
|
name: str
|
|
|
|
# Human-readable name shown in UI
|
|
display_name: str
|
|
|
|
# Which device categories this action supports. Empty list = all categories.
|
|
compatible_categories: list[str] = []
|
|
|
|
# Parameter definitions for UI form generation
|
|
params_schema: list[dict] = []
|
|
|
|
@abstractmethod
|
|
async def execute(
|
|
self,
|
|
device: Device,
|
|
obj: Object,
|
|
params: dict,
|
|
emit: Emitter,
|
|
) -> ActionResult: ...
|
|
|
|
def is_compatible(self, category: str) -> bool:
|
|
return not self.compatible_categories or category in self.compatible_categories
|