| from __future__ import annotations |
|
|
| from abc import ABC, abstractmethod |
| from typing import Any, Dict, List |
|
|
| from .context import LogErrorLevel, PluginContext |
|
|
|
|
| class Plugin(ABC): |
| """ |
| base class for all plugins |
| |
| the instance of the plugin is a callable object, which is the entry point for |
| the execution of the plugin function. The execution context and |
| the configuration of the plugin are passed to the plugin instance when it is created. |
| """ |
|
|
| def __init__(self, name: str, ctx: PluginContext, config: Dict[str, Any]) -> None: |
| """ |
| create a plugin instance, this method will be called by the runtime |
| |
| :param name: the name of the plugin |
| :param ctx: the execution context of the plugin |
| :param config: the configuration of the plugin |
| """ |
| super().__init__() |
| self.name: str = name |
| self.ctx: PluginContext = ctx |
| self.config: Dict[str, Any] = config |
|
|
| @abstractmethod |
| def __call__(self, *args: List[Any], **kwargs: Dict[str, Any]) -> Any: |
| """ |
| entry point for the execution of the plugin function |
| """ |
|
|
| def log(self, level: LogErrorLevel, message: str) -> None: |
| """log a message from the plugin""" |
| self.ctx.log(level, "Plugin-" + self.name, message) |
|
|
| def get_env(self, variable_name: str) -> str: |
| """get an environment variable from the context""" |
| return self.ctx.get_env(self.name, variable_name) |
|
|
