1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
|
from __future__ import annotations
import os
from typing import Union
import yaml
from pydantic import BaseModel
from autogpt.logs import logger
from autogpt.plugins.plugin_config import PluginConfig
class PluginsConfig(BaseModel):
"""Class for holding configuration of all plugins"""
plugins: dict[str, PluginConfig]
def __repr__(self):
return f"PluginsConfig({self.plugins})"
def get(self, name: str) -> Union[PluginConfig, None]:
return self.plugins.get(name)
def is_enabled(self, name) -> bool:
plugin_config = self.plugins.get(name)
return plugin_config is not None and plugin_config.enabled
@classmethod
def load_config(
cls,
plugins_config_file: str,
plugins_denylist: list[str],
plugins_allowlist: list[str],
) -> "PluginsConfig":
empty_config = cls(plugins={})
try:
config_data = cls.deserialize_config_file(
plugins_config_file,
plugins_denylist,
plugins_allowlist,
)
if type(config_data) != dict:
logger.error(
f"Expected plugins config to be a dict, got {type(config_data)}, continuing without plugins"
)
return empty_config
return cls(plugins=config_data)
except BaseException as e:
logger.error(
f"Plugin config is invalid, continuing without plugins. Error: {e}"
)
return empty_config
@classmethod
def deserialize_config_file(
cls,
plugins_config_file: str,
plugins_denylist: list[str],
plugins_allowlist: list[str],
) -> dict[str, PluginConfig]:
if not os.path.exists(plugins_config_file):
logger.warn("plugins_config.yaml does not exist, creating base config.")
cls.create_empty_plugins_config(
plugins_config_file,
plugins_denylist,
plugins_allowlist,
)
with open(plugins_config_file, "r") as f:
plugins_config = yaml.load(f, Loader=yaml.FullLoader)
plugins = {}
for name, plugin in plugins_config.items():
if type(plugin) == dict:
plugins[name] = PluginConfig(
name=name,
enabled=plugin.get("enabled", False),
config=plugin.get("config", {}),
)
elif type(plugin) == PluginConfig:
plugins[name] = plugin
else:
raise ValueError(f"Invalid plugin config data type: {type(plugin)}")
return plugins
@staticmethod
def create_empty_plugins_config(
plugins_config_file: str,
plugins_denylist: list[str],
plugins_allowlist: list[str],
):
"""Create an empty plugins_config.yaml file. Fill it with values from old env variables."""
base_config = {}
logger.debug(f"Legacy plugin denylist: {plugins_denylist}")
logger.debug(f"Legacy plugin allowlist: {plugins_allowlist}")
# Backwards-compatibility shim
for plugin_name in plugins_denylist:
base_config[plugin_name] = {"enabled": False, "config": {}}
for plugin_name in plugins_allowlist:
base_config[plugin_name] = {"enabled": True, "config": {}}
logger.debug(f"Constructed base plugins config: {base_config}")
logger.debug(f"Creating plugin config file {plugins_config_file}")
with open(plugins_config_file, "w+") as f:
f.write(yaml.dump(base_config))
return base_config
|
