1#!/usr/bin/env python3
2"""
3
4"""
5# Imports:
6from __future__ import annotations
7
8# ##-- stdlib imports
9import datetime
10import enum
11import functools as ftz
12import importlib
13import itertools as itz
14import logging as logmod
15import pathlib as pl
16import re
17import time
18import types
19from importlib.metadata import entry_points
20from uuid import UUID, uuid1
21
22# ##-- end stdlib imports
23
24# ##-- 3rd party imports
25from jgdv import Proto
26from jgdv.structs.chainguard import ChainGuard
27
28# ##-- end 3rd party imports
29
30# ##-- 1st party imports
31import doot
32
33# ##-- end 1st party imports
34
35# ##-- types
36# isort: off
37import abc
38import collections.abc
39from typing import TYPE_CHECKING, cast, assert_type, assert_never
40from typing import Generic, NewType
41# Protocols:
42from typing import Protocol, runtime_checkable
43# Typing Decorators:
44from typing import no_type_check, final, override, overload
45
46if TYPE_CHECKING:
47 from jgdv import Maybe
48 from typing import Final
49 from typing import ClassVar, Any, LiteralString
50 from typing import Never, Self, Literal
51 from typing import TypeGuard
52 from collections.abc import Iterable, Iterator, Callable, Generator
53 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
54
55##--|
56from doot.cmds._interface import Command_p
57from ._interface import CommandLoader_p
58# isort: on
59# ##-- end types
60
61##-- logging
62logging = logmod.getLogger(__name__)
63##-- end logging
64
[docs]
65@Proto(CommandLoader_p)
66class CommandLoader:
67 """
68 Default Command loaded. using the loaded plugins,
69 selects "command", calls load on each entry point, and if the obj returned is a subclass of Command_p,
70 instantiates it
71 """
72
[docs]
73 def setup(self, plugins, extra:Maybe[list|dict|ChainGuard]=None) -> Self:
74 self.cmd_plugins : list[EntryPoint] = plugins.get("command", [])
75 self.cmds = {}
76
77 match extra:
78 case None:
79 self.extra = []
80 case list():
81 self.extra = extra
82 case dict():
83 self.extra = ChainGuard(extra).on_fail([]).tasks()
84 case ChainGuard():
85 self.extra = extra.on_fail([]).tasks()
86
87 return self
88
[docs]
89 def load(self) -> ChainGuard[Command_p]:
90 logging.debug("---- Loading Commands")
91 for cmd_point in self.cmd_plugins:
92 try:
93 logging.debug("Loading Cmd: %s", cmd_point.name)
94 # load the plugins
95 cmd = cmd_point.load()
96 if not isinstance(cmd, Command_p):
97 raise TypeError("Not a Command_p", cmd)
98
99 self.cmds[cmd_point.name] = cmd()
100 self.cmds[cmd_point.name]._name = cmd_point.name
101 except Exception as err:
102 raise doot.errors.PluginLoadError("Attempted to load a non-command: %s : %s", cmd_point, err) from err
103
104 return ChainGuard(self.cmds)