1
2#!/usr/bin/env python3
3"""
4
5"""
6# Imports:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import datetime
11import enum
12import functools as ftz
13import importlib
14import itertools as itz
15import logging as logmod
16import re
17import time
18import types
19from collections import defaultdict
20from importlib.metadata import EntryPoint, entry_points
21from uuid import UUID, uuid1
22
23# ##-- end stdlib imports
24
25# ##-- 3rd party imports
26from jgdv import Proto
27from jgdv.structs.chainguard import ChainGuard
28
29# ##-- end 3rd party imports
30
31# ##-- 1st party imports
32import doot
33from . import _interface as API # noqa: N812
34# ##-- end 1st party imports
35
36# ##-- types
37# isort: off
38import abc
39import collections.abc
40from typing import TYPE_CHECKING, cast, assert_type, assert_never
41from typing import Generic, NewType
42# Protocols:
43from typing import Protocol, runtime_checkable
44# Typing Decorators:
45from typing import no_type_check, final, override, overload
46
47from ._interface import PluginLoader_p
48
49if TYPE_CHECKING:
50 import pathlib as pl
51 from jgdv import Maybe
52 from typing import Final
53 from typing import ClassVar, Any, LiteralString
54 from typing import Never, Self, Literal
55 from typing import TypeGuard
56 from collections.abc import Iterable, Iterator, Callable, Generator
57 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
58
59# isort: on
60# ##-- end types
61
62##-- logging
63logging = logmod.getLogger(__name__)
64##-- end logging
65
66##--| vars
67skip_default_plugins : Final[bool] = doot.config.on_fail(False).startup.skip_default_plugins() # noqa: FBT003
68skip_plugin_search : Final[bool] = doot.config.on_fail(False).startup.skip_plugin_search() # noqa: FBT003
69env_plugins : Final[dict] = doot.config.on_fail({}).startup.plugins(wrapper=dict) # type: ignore[arg-type]
70
71# Constants:
72## The plugin types to search for:
73frontend_plugins : Final[list] = doot.constants.entrypoints.FRONTEND_PLUGIN_TYPES # type: ignore[attr-defined]
74backend_plugins : Final[list] = doot.constants.entrypoints.BACKEND_PLUGIN_TYPES # type: ignore[attr-defined]
75plugin_types : Final[set] = set(frontend_plugins + backend_plugins)
76
77cmd_loader_key : Final[str] = doot.constants.entrypoints.DEFAULT_COMMAND_LOADER_KEY # type: ignore[attr-defined]
78task_loader_key : Final[str] = doot.constants.entrypoints.DEFAULT_TASK_LOADER_KEY # type: ignore[attr-defined]
79PLUGIN_PREFIX : Final[str] = doot.constants.entrypoints.PLUGIN_TOML_PREFIX # type: ignore[attr-defined]
80DEFAULT_CMD_LOADER : Final[str] = doot.constants.entrypoints.DEFAULT_COMMAND_LOADER # type: ignore[attr-defined]
81DEFAULT_TASK_LOADER : Final[str] = doot.constants.entrypoints.DEFAULT_TASK_LOADER # type: ignore[attr-defined]
82DEFAULT_TASK_GROUP : Final[str] = doot.constants.names.DEFAULT_TASK_GROUP # type: ignore[attr-defined]
83
84# Other
85TOML_SUFFIX : Final[str] = ".toml"
86
87##--| util
[docs]
88def build_entry_point (x:str, y:str, z:str) -> EntryPoint:
89 """ Make an EntryPoint """
90 if z not in plugin_types:
91 raise doot.errors.PluginError("Plugin Type Not Found: %s : %s", z, (x, y))
92 group = f"{PLUGIN_PREFIX}.{z}"
93 return EntryPoint(name=x, value=y, group=group)
94
[docs]
95@Proto(PluginLoader_p)
96class PluginLoader:
97 """
98 Load doot plugins from the system, to choose from with doot.toml or cli args
99 TODO singleton?
100 """
101
[docs]
102 def setup(self, extra_config:Maybe[dict|ChainGuard]=None) -> Self:
103 self.plugins : dict = defaultdict(list)
104 match extra_config:
105 case None:
106 self.extra_config = ChainGuard({})
107 case dict():
108 self.extra_config = ChainGuard(extra_config)
109 case ChainGuard():
110 self.extra_config = extra_config
111
112 return self
113
[docs]
114 def load(self) -> ChainGuard[EntryPoint]: # type: ignore[type-arg]
115 """
116 use entry_points(group="doot")
117 add to the config ChainGuard
118 """
119 logging.debug("---- Loading Plugins: %s", doot.constants.entrypoints.PLUGIN_TOML_PREFIX) # type: ignore[attr-defined]
120 try:
121 self._load_system_plugins()
122 except Exception as err:
123 raise doot.errors.PluginError("Failed to load system wide plugins: %s", err) from err
124
125 try:
126 self._load_from_toml()
127 except Exception as err:
128 raise doot.errors.PluginError("Failed to load toml specified plugins: %s", err) from err
129
130 try:
131 self._load_extra_plugins()
132 except Exception as err:
133 raise doot.errors.PluginError("Failed to load command line/dooter specified plugins: %s", err) from err
134
135 try:
136 self._append_defaults()
137 except Exception as err:
138 raise doot.errors.PluginError("Failed to load plugin defaults: %s", err) from err
139
140 logging.debug("Found %s plugins", len(self.plugins))
141 loaded = ChainGuard(self.plugins)
142 return loaded
143
[docs]
144 def _load_system_plugins(self) -> None:
145 plugin_group : str
146 entry_point : EntryPoint
147 if skip_plugin_search:
148 return
149
150 logging.info("-- Searching environment for plugins, skip with `skip_plugin_search` in config")
151 for plugin_type in plugin_types:
152 try:
153 plugin_group = f"{PLUGIN_PREFIX}.{plugin_type}"
154 # Load env wide entry points
155 for entry_point in entry_points(group=plugin_group):
156 self.plugins[plugin_type].append(entry_point)
157 except Exception as err:
158 raise doot.errors.PluginError("Plugin Failed to Load: %s : %s : %s", plugin_group, entry_point, err) from err
159
[docs]
160 def _load_from_toml(self) -> None:
161 logging.info("-- Loading Plugins from Toml")
162 # load config entry points
163 for cmd_group, vals in env_plugins.items():
164 if cmd_group not in plugin_types:
165 logging.warning("Unknown plugin type found in config: %s", cmd_group)
166 continue
167
168 if not isinstance(vals, ChainGuard|dict):
169 logging.warning("Toml specified plugins need to be a dict of (cmdName : class): %s ", cmd_group)
170 continue
171
172 for name, cls in vals.items():
173 logging.debug("Creating Plugin Entry Point: %s : %s", cmd_group, name)
174 ep = build_entry_point(name, cls, cmd_group)
175 self.plugins[cmd_group].append(ep)
176
191
[docs]
192 def _append_defaults(self) -> None:
193 if skip_default_plugins:
194 return
195
196 logging.info("-- Loading Default Plugin Aliases")
197 self.plugins[cmd_loader_key].append(build_entry_point(cmd_loader_key, DEFAULT_CMD_LOADER, cmd_loader_key))
198 self.plugins[task_loader_key].append(build_entry_point(task_loader_key, DEFAULT_TASK_LOADER, task_loader_key))
199
200 for group, vals in doot.aliases.items():
201 logging.debug("Loading aliases: %s (%s)", group, len(vals))
202 defined = {x.name for x in self.plugins[group]}
203 defaults = {x : build_entry_point(x, y, group) for x,y in vals.items() if x not in defined}
204 self.plugins[group] += defaults.values()