Coverage for src/duty/collection.py: 98.99%
79 statements
« prev ^ index » next coverage.py v7.6.3, created at 2024-10-17 17:18 +0200
1"""Module containing all the logic."""
3from __future__ import annotations
5import inspect
6import sys
7from copy import deepcopy
8from importlib import util as importlib_util
9from typing import Any, Callable, ClassVar, Union
11from duty.context import Context
13DutyListType = list[Union[str, Callable, "Duty"]]
14default_duties_file = "duties.py"
17class Duty:
18 """The main duty class."""
20 default_options: ClassVar[dict[str, Any]] = {}
22 def __init__(
23 self,
24 name: str,
25 description: str,
26 function: Callable,
27 collection: Collection | None = None,
28 aliases: set | None = None,
29 pre: DutyListType | None = None,
30 post: DutyListType | None = None,
31 opts: dict[str, Any] | None = None,
32 ) -> None:
33 """Initialize the duty.
35 Parameters:
36 name: The duty name.
37 description: The duty description.
38 function: The duty function.
39 collection: The collection on which to attach this duty.
40 aliases: A list of aliases for this duty.
41 pre: A list of duties to run before this one.
42 post: A list of duties to run after this one.
43 opts: Options used to create the context instance.
44 """
45 self.name = name
46 self.description = description
47 self.function = function
48 self.aliases = aliases or set()
49 self.pre = pre or []
50 self.post = post or []
51 self.options = opts or self.default_options
52 self.options_override: dict = {}
54 self.collection: Collection | None = None
55 if collection:
56 collection.add(self)
58 @property
59 def context(self) -> Context:
60 """Return a new context instance.
62 Returns:
63 A new context instance.
64 """
65 return Context(self.options, self.options_override)
67 def run(self, *args: Any, **kwargs: Any) -> None:
68 """Run the duty.
70 This is just a shortcut for `duty(duty.context, *args, **kwargs)`.
72 Parameters:
73 args: Positional arguments passed to the function.
74 kwargs: Keyword arguments passed to the function.
75 """
76 self(self.context, *args, **kwargs)
78 def run_duties(self, context: Context, duties_list: DutyListType) -> None:
79 """Run a list of duties.
81 Parameters:
82 context: The context to use.
83 duties_list: The list of duties to run.
85 Raises:
86 RuntimeError: When a duty name is given to pre or post duties.
87 Indeed, without a parent collection, it is impossible
88 to find another duty by its name.
89 """
90 for duty_item in duties_list:
91 if callable(duty_item):
92 # Item is a proper duty, or a callable: run it.
93 duty_item(context)
94 elif isinstance(duty_item, str):
95 # Item is a reference to a duty.
96 if self.collection is None:
97 raise RuntimeError(f"Can't find duty by name without a collection ({duty_item})")
98 # Get the duty and run it.
99 self.collection.get(duty_item)(context)
101 def __call__(self, context: Context, *args: Any, **kwargs: Any) -> None:
102 """Run the duty function.
104 Parameters:
105 context: The context to use.
106 args: Positional arguments passed to the function.
107 kwargs: Keyword arguments passed to the function.
108 """
109 self.run_duties(context, self.pre)
110 self.function(context, *args, **kwargs)
111 self.run_duties(context, self.post)
114class Collection:
115 """A collection of duties.
117 Attributes:
118 path: The path to the duties file.
119 duties: The list of duties.
120 aliases: A dictionary of aliases pointing to their respective duties.
121 """
123 def __init__(self, path: str = default_duties_file) -> None:
124 """Initialize the collection.
126 Parameters:
127 path: The path to the duties file.
128 """
129 self.path = path
130 self.duties: dict[str, Duty] = {}
131 self.aliases: dict[str, Duty] = {}
133 def clear(self) -> None:
134 """Clear the collection."""
135 self.duties.clear()
136 self.aliases.clear()
138 def names(self) -> list[str]:
139 """Return the list of duties names and aliases.
141 Returns:
142 The list of duties names and aliases.
143 """
144 return list(self.duties.keys()) + list(self.aliases.keys())
146 def get(self, name_or_alias: str) -> Duty:
147 """Get a duty by its name or alias.
149 Parameters:
150 name_or_alias: The name or alias of the duty.
152 Returns:
153 A duty.
154 """
155 try:
156 return self.duties[name_or_alias]
157 except KeyError:
158 return self.aliases[name_or_alias]
160 def format_help(self) -> str:
161 """Format a message listing the duties.
163 Returns:
164 A string listing the duties and their summary.
165 """
166 lines = []
167 # 20 makes the summary aligned with options description
168 longest_name = max(*(len(name) for name in self.duties), 20)
169 for name, duty in self.duties.items():
170 description = duty.description.split("\n")[0]
171 lines.append(f"{name:{longest_name}} {description}")
172 return "\n".join(lines)
174 def add(self, duty: Duty) -> None:
175 """Add a duty to the collection.
177 Parameters:
178 duty: The duty to add.
179 """
180 if duty.collection is not None:
181 # we must copy the duty to be able to add it
182 # in multiple collections
183 duty = deepcopy(duty)
184 duty.collection = self
185 self.duties[duty.name] = duty
186 for alias in duty.aliases:
187 self.aliases[alias] = duty
189 def load(self, path: str | None = None) -> None:
190 """Load duties from a Python file.
192 Parameters:
193 path: The path to the Python file to load.
194 Uses the collection's path by default.
195 """
196 path = path or self.path
197 spec = importlib_util.spec_from_file_location("duty.duties", path)
198 if spec: 198 ↛ exitline 198 didn't return from function 'load' because the condition on line 198 was always true
199 duties = importlib_util.module_from_spec(spec)
200 sys.modules["duty.duties"] = duties
201 spec.loader.exec_module(duties) # type: ignore[union-attr]
202 declared_duties = inspect.getmembers(duties, lambda member: isinstance(member, Duty))
203 for _, duty in declared_duties:
204 self.add(duty)