# pylint: disable=W0613
"""Plugin utils module for core plugin functionality."""
import os
import re
from typing import Iterator, Optional, Union
import yaml
from smarter.apps.account.models import Account, User, UserProfile
from smarter.apps.account.utils import (
get_cached_smarter_admin_user_profile,
)
from smarter.apps.llmtool.manifest.controller import PluginController
from smarter.apps.llmtool.models import PluginDataValueError, PluginMeta
from smarter.common.const import PYTHON_ROOT
from smarter.lib import logging
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
from smarter.lib.typing import SmarterDictType
from .base import PluginBase
[docs]
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.PLUGIN_LOGGING)
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
[docs]
class Plugins:
"""
A class for managing and interacting with multiple plugins.
This class provides methods to retrieve, serialize, and work with plugins associated with a user and account. It loads plugins using metadata and controllers, and exposes their data in dictionary and JSON formats.
:param user: The user for whom plugins are loaded.
:type user: User
:param account: The account context for plugin retrieval.
:type account: Account
.. note::
A plugin that cannot be loaded, for example one whose plugin data
is missing, is logged and skipped.
.. seealso::
:class:`PluginController` for plugin instantiation.
:class:`PluginMeta` for plugin metadata.
**Example usage**::
plugins = Plugins(user=my_user, account=my_account)
plugin_dicts = plugins.data
plugin_json = plugins.to_json()
"""
account: Optional[Account] = None
user_profile: Optional[UserProfile] = None
plugins: list[PluginBase] = []
[docs]
def __init__(self, user: User, account: Account):
self.plugins = []
self.account = account or UserProfile.get_cached_object(user=user).account
self.user_profile = UserProfile.get_cached_object(user=user, account=account)
# plugins for this user profile, followed by those of the smarter admin user profile
smarter_admin_user_profile = get_cached_smarter_admin_user_profile()
for user_profile in (self.user_profile, smarter_admin_user_profile):
for plugin_meta in PluginMeta.objects.filter(user_profile=user_profile):
plugin = self.load_plugin(plugin_meta=plugin_meta, user_profile=user_profile)
if plugin is not None:
self.plugins.append(plugin)
def __len__(self) -> int:
"""Return the number of loaded plugins, ready or not."""
return len(self.plugins)
def __iter__(self) -> Iterator[PluginBase]:
"""Iterate over the loaded plugins, ready or not."""
return iter(self.plugins)
def __repr__(self) -> str:
return f"<{self.__class__.__name__} user_profile={self.user_profile} count={len(self)}>"
[docs]
@staticmethod
def load_plugin(plugin_meta: PluginMeta, user_profile: UserProfile) -> Optional[PluginBase]:
"""
Load a plugin, or log and skip it if it cannot be loaded.
One malformed plugin, for example one whose plugin data is missing,
should not prevent the others from being listed.
:param plugin_meta: The plugin's PluginMeta.
:type plugin_meta: PluginMeta
:param user_profile: The plugin's owner.
:type user_profile: Optional[UserProfile]
:returns: The plugin, or None if it cannot be loaded.
:rtype: Optional[PluginBase]
"""
try:
plugin_controller = PluginController(
user_profile=user_profile,
plugin_meta=plugin_meta,
)
if not plugin_controller or not plugin_controller.plugin:
raise PluginDataValueError(
f"PluginController could not be created for plugin_id: {plugin_meta.id}, user_profile: {user_profile}" # type: ignore[arg-type]
)
return plugin_controller.plugin
# pylint: disable=W0718
except Exception as e:
base_logger.warning(
"Plugins.load_plugin() skipping plugin %s (id: %s) for %s, which cannot be loaded: %s",
plugin_meta.name,
plugin_meta.id, # type: ignore[reportAttributeAccessIssue]
user_profile,
e,
)
return None
@property
def data(self) -> list[SmarterDictType]:
"""Return a list of plugins in dictionary format."""
retval = []
for plugin in self.plugins:
if plugin.ready:
retval.append(plugin.data)
return retval
[docs]
def to_json(self) -> list[SmarterDictType]:
"""Return a list of plugins in JSON format."""
retval = []
for plugin in self.plugins:
if plugin.ready:
retval.append(plugin.to_json())
return retval
[docs]
class PluginExample:
"""
A class for loading and working with built-in YAML-based plugin examples.
This class reads plugin example files in YAML format, parses their contents, and exposes metadata and serialization methods for inspection and testing.
:param filepath: The directory path containing the YAML file.
:type filepath: str
:param filename: The name of the YAML file to load.
:type filename: str
.. seealso::
:class:`PluginExamples` for managing collections of plugin examples.
**Example usage**::
example = PluginExample(filepath="/path/to/examples", filename="my_plugin.yaml")
print(example.name)
print(example.to_yaml())
print(example.to_json())
"""
_filename: Optional[str]
_json: Optional[Union[list, SmarterDictType]]
_yaml: Optional[str]
[docs]
def __init__(self, filepath: str, filename: str):
"""Initialize the class from a yaml file."""
with open(os.path.join(filepath, filename), encoding="utf-8") as file:
self._yaml = file.read()
self._json = yaml.safe_load(self._yaml)
self._filename = filename
def __repr__(self) -> str:
return f"<{self.__class__.__name__} filename={self.filename} name={self.name}>"
@property
def filename(self) -> Optional[str]:
"""Return the name of the plugin."""
return self._filename
@property
def name(self) -> Optional[str]:
"""Return the name of the plugin."""
try:
retval = self._json["metadata"]["name"] if isinstance(self._json, dict) else None
except KeyError:
logger.warning("PluginExample: %s is malformed and has no metadata.name", self.filename)
retval = self.convert_filename()
return retval
[docs]
def to_yaml(self) -> Optional[str]:
"""Return the plugin as a yaml string."""
return self._yaml
# TODO: this fails on Plugin.create() due to missing tags
# django.core.exceptions.ValidationError: ["Invalid data: missing meta_data['tags']"]
[docs]
def to_json(self) -> Optional[Union[SmarterDictType, list]]:
"""Return the plugin as a dictionary."""
return self._json
[docs]
def convert_filename(self) -> Optional[str]:
"""Convert the filename to the desired format."""
if not isinstance(self.filename, str):
return self.filename
try:
filename = os.path.splitext(self.filename)[0] # Remove the file extension
name = re.sub(r"[-_]", " ", filename) # Replace hyphens and underscores with spaces
name = name.title().replace(" ", "") # Capitalize each word and remove spaces
return name
# pylint: disable=broad-except
except Exception as e:
logger.error("PluginExample: %s failed to convert filename: %s", self.filename, e)
return self.filename
[docs]
class PluginExamples:
"""
A class for managing a collection of :class:`PluginExample` instances.
This class loads all YAML-based plugin examples from a specified directory, providing access to the collection and utility methods for counting and retrieving examples.
:param args: Optional positional arguments (unused).
:type args: tuple
:param kwargs: Optional keyword arguments (unused).
:type kwargs: dict
.. note::
Only files ending with ``.yaml`` in the plugins path are loaded as examples.
.. tip::
Use :meth:`count` to get the number of loaded plugin examples, and the :meth:`plugins` property to access the list.
.. seealso::
:class:`PluginExample` for individual example details.
**Example usage**::
examples = PluginExamples()
print(examples.count())
for example in examples.plugins:
print(example.filename, example.name)
"""
_plugin_examples: list[PluginExample] = []
HERE = os.path.abspath(os.path.dirname(__file__))
PLUGINS_PATH = os.path.join(PYTHON_ROOT, "smarter", "apps", "llmtool", "data", "sample-plugins")
[docs]
def __init__(self, *args, **kwargs):
"""Initialize the class."""
self._plugin_examples = []
# sorted, because os.listdir() order depends on the filesystem, and the
# api contract's committed LLMClient example lists these names.
for file in sorted(os.listdir(self.PLUGINS_PATH)):
if file.endswith(".yaml"):
plugin_example = PluginExample(filepath=self.PLUGINS_PATH, filename=file)
self._plugin_examples.append(plugin_example)
def __len__(self) -> int:
"""Return the number of plugin examples."""
return len(self._plugin_examples)
def __iter__(self) -> Iterator[PluginExample]:
"""Iterate over the plugin examples."""
return iter(self._plugin_examples)
[docs]
def count(self) -> int:
"""Return the number of plugins."""
return len(self._plugin_examples)
@property
def plugins(self) -> list[PluginExample]:
"""Return a list of plugins in dictionary format."""
return self._plugin_examples