base_plugin — BasePlugin, PluginConfig, Exceptions

Overview

Every plugin in the ecosystem inherits from BasePlugin. It centralizes everything that should not be reimplemented per-plugin: input validation dispatch, timeout enforcement, retry with backoff, result caching, execution statistics, and health scoring. Subclasses (directly, or via one of the four plugin types) only need to implement the actual plugin logic.

Exceptions

class PluginError(Exception): ...
class PluginTimeoutError(PluginError): ...
class PluginValidationError(PluginError): ...
class PluginPermissionError(PluginError): ...

All four are plain exception subclasses with no additional attributes beyond the inherited args.

PluginConfig

PluginConfig(
    enabled: bool = True,
    timeout: float = 30.0,
    max_retries: int = 3,
    cache_results: bool = False,
    cache_ttl_sec: int = 300,
    custom: Dict[str, Any] = None,
)

Per-instance runtime configuration. Constructed once per plugin instance (or shared as a default_config across every plugin discovered by a PluginLoader).

Parameters

Raises

• ValueError — if timeout <= 0 or max_retries < 0.

BasePlugin

class BasePlugin(ABC):
    METADATA: PluginMetadata  # must be set by subclass at class level

    def __init__(self, config: Optional[PluginConfig] = None) -> None: ...

Constructor behavior

1. Checks hasattr(self.__class__, "METADATA") — raises TypeError immediately if absent.
2. Copies self.metadata, self.name from the class-level METADATA.
3. Sets self.config = config or PluginConfig().
4. Sets self.status = PluginStatus.DISABLED.
5. Initializes execution counters (_exec_total, _exec_success, _exec_failure), timing stats (_time_total, _time_min = float("inf"), _time_max, _time_last), timestamps (initialized_at, last_executed_at, last_error_at — all None initially), a capped error journal (_error_journal, max 50 entries), a result cache (_result_cache, _cache_ts), and an asyncio.Lock() used to serialize concurrent safe_execute() calls on the same instance.

Raises

• TypeError — if the subclass has no class-level METADATA attribute.

Abstract interface (subclasses must implement)

Optional overrides

Public execution surface

BasePlugin.safe_execute

async def safe_execute(self, input_data: Dict[str, Any], context: ExecutionContext) -> Any

Internal workflow:

1. Raise PluginError if not self.config.enabled.
2. Raise PluginError if self.status == PluginStatus.ERROR (includes self.error_msg in the message).
3. validated = await self.validate(input_data).
4. If config.cache_results: look up a SHA-256 hash of the sorted-JSON-serialized validated input in the local cache; return the cached value immediately on a hit within cache_ttl_sec.
5. Acquire self._lock (serializes concurrent calls to this specific instance).
6. Run execute(validated, context) inside asyncio.wait_for(timeout=self.config.timeout).
7. On success: record stats via _record_success, populate the cache if enabled, return the result.
8. On asyncio.TimeoutError: record failure, raise PluginTimeoutError.
9. On PluginError/PluginValidationError: re-raise unchanged.
10. On any other Exception: record failure, raise PluginError wrapping the original as __cause__.

Raises: PluginError, PluginTimeoutError, PluginValidationError (propagated from validate() or execute()).

Thread safety: Not thread-safe across OS threads. Async-safe for concurrent coroutines on the same event loop — the internal asyncio.Lock serializes overlapping calls to the same instance, so two concurrent await plugin.safe_execute(...) calls on one instance run one after another, not in parallel. Multiple plugin instances execute fully concurrently.

Side effects: mutates self.status (possibly to ERROR after repeated failures), execution counters, _error_journal, _result_cache; emits logger.debug/logger.warning log records.

BasePlugin.safe_execute_with_retry

async def safe_execute_with_retry(self, input_data: Dict[str, Any], context: ExecutionContext) -> Any

Calls safe_execute() up to config.max_retries + 1 times. On each PluginError (including PluginTimeoutError, a subclass), sleeps 2 ** attempt seconds (exponential backoff: 1s, 2s, 4s, …) before the next attempt, except after the final attempt. Re-raises the last exception if all attempts are exhausted.

Raises: the last PluginError encountered, or a generic PluginError("All retries exhausted.") if somehow no exception was captured (defensive fallback; not reachable in normal control flow since the loop always either returns or captures an exception).

Stats & health

get_stats() -> Dict[str, Any]

Returns a nested dict: name, version, status, enabled, an execution block (total/success/failure/success_rate as a formatted percentage string), a timing block (total_s/avg_s/min_s/max_s/last_s, all rounded to 3 decimals; min_s is None if no successful execution has occurred), a timestamps block (ISO-formatted or None), error_journal_size, and current_error.

get_health() -> Dict[str, Any]

Computes a label/score pair from the success rate:

Also returns status and enabled.

reset_stats() -> None

Zeroes every counter and clears the error journal. Does not reset status, initialized_at, or the result cache.

Internal helpers (private, documented for completeness)

• _record_success(t0) / _record_failure(t0, msg) — update counters and timestamps; _record_failure also appends to _error_journal (trimming to the last 50 entries) and force-transitions status → PluginStatus.ERROR once _exec_failure > 10 and the failure rate exceeds 50%.
• _cache_key(data) — SHA-256 hex digest of json.dumps(data, sort_keys=True, default=str).
• _cache_get(data) / _cache_set(data, result) — TTL-based cache read/write keyed by _cache_key.

[!WARNING] Known inconsistency in the source _record_success and _record_failure compute elapsed time using datetime.utcnow(), while safe_execute captures t0 using datetime.now(UTC). Both are naive vs. timezone-aware datetime objects respectively in some Python versions; subtracting them can behave unexpectedly depending on your Python version's handling of aware/naive datetime arithmetic. This is a direct observation from the source, not a claim about a specific failure mode — verify timing behavior in your target Python version if precise timing matters.

__repr__

f"{self.__class__.__name__}(name={self.name!r}, version={self.metadata.version!r}, status={self.status.value!r})"

Complete example

from zerda.plugins import BasePlugin, PluginConfig, PluginMetadata, SchemaProperty, ExecutionContext

class WordCountPlugin(BasePlugin):
    METADATA = PluginMetadata(
        name="word_count",
        version="1.0.0",
        author="docs-example",
        description="Counts words in the given text.",
        input_schema=[
            SchemaProperty(name="text", type="string", description="Text to count words in"),
        ],
    )

    async def initialize(self) -> bool:
        return True

    async def execute(self, input_data, context: ExecutionContext) -> dict:
        return {"count": len(input_data["text"].split())}

    async def cleanup(self) -> None:
        pass

plugin = WordCountPlugin(config=PluginConfig(timeout=5.0, cache_results=True))
await plugin.initialize()
plugin.status = __import__("plugins").PluginStatus.ENABLED  # normally done by PluginManager
result = await plugin.safe_execute({"text": "hello world"}, ExecutionContext())
print(result)          # {"count": 2}
print(plugin.get_stats()["execution"]["success_rate"])   # "100.0%"

[!NOTE] In this minimal example status is set manually. In normal usage, PluginManager._initialize_plugin sets status = PluginStatus.ENABLED after a successful initialize() call — you should not need to set it by hand when going through PluginManager.

Related APIs

• PluginMetadata — required class-level attribute
• ExecutionContext — second argument to execute()
• PluginManager — the usual caller of safe_execute / safe_execute_with_retry
• InputSanitizer — schema-level validation performed before safe_execute is reached, when invoked through PluginManager.execute()

Version history

plugins/__init__.py declares __version__ = "2.0.0" for the package as a whole. No per-module or per-class version/changelog annotations exist in the source, so class-level version history cannot be determined.