Coverage for src / graphable / graphable.py: 99%
176 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-16 21:32 +0000
1from collections import deque
2from logging import getLogger
3from typing import Any, Protocol, Self, cast, runtime_checkable
4from weakref import WeakSet
6from .errors import GraphCycleError
8logger = getLogger(__name__)
11@runtime_checkable
12class GraphObserver(Protocol):
13 """Protocol for objects that need to be notified of node changes."""
15 def _invalidate_cache(self) -> None: ...
18class Graphable[T]:
19 """
20 A generic class representing a node in a graph that can track dependencies and dependents.
22 Type Parameters:
23 T: The type of the reference object this node holds.
24 """
26 def __init__(self, reference: T):
27 """
28 Initialize a Graphable node.
30 Args:
31 reference (T): The underlying object this node represents.
32 """
33 self._dependents: dict[Graphable[Any], dict[str, Any]] = {}
34 self._depends_on: dict[Graphable[Any], dict[str, Any]] = {}
35 self._reference: T = reference
36 self._tags: set[str] = set()
37 self._observers: WeakSet[GraphObserver] = WeakSet()
38 self._duration: float = 0.0
39 self._status: str = "pending"
40 logger.debug(f"Created Graphable node for reference: {reference}")
42 def _notify_change(self) -> None:
43 """Notify all observers that this node has changed."""
44 for observer in list(self._observers):
45 observer._invalidate_cache()
47 def _register_observer(self, observer: GraphObserver) -> None:
48 """Register an observer to be notified of changes."""
49 self._observers.add(observer)
51 def _unregister_observer(self, observer: GraphObserver) -> None:
52 """Unregister an observer."""
53 self._observers.discard(observer)
55 @property
56 def duration(self) -> float:
57 """Get the duration of this node."""
58 return self._duration
60 @duration.setter
61 def duration(self, value: float) -> None:
62 """Set the duration of this node."""
63 self._duration = value
64 self._notify_change()
66 @property
67 def status(self) -> str:
68 """Get the status of this node."""
69 return self._status
71 @status.setter
72 def status(self, value: str) -> None:
73 """Set the status of this node."""
74 self._status = value
75 self._notify_change()
77 def __eq__(self, other: object) -> bool:
78 """
79 Check if this node is equal to another.
80 Nodes are considered equal if they are the same object.
81 """
82 return self is other
84 def __hash__(self) -> int:
85 """
86 Restore default hash behavior (identity-based).
87 """
88 return id(self)
90 def __lt__(self, other: object) -> bool:
91 """
92 Check if this node is 'less than' another.
93 Defined as: this node is a proper ancestor of the other node.
94 """
95 if not isinstance(other, Graphable):
96 return NotImplemented
98 return self.find_path(other) is not None
100 def __le__(self, other: object) -> bool:
101 """
102 Check if this node is 'less than or equal' to another.
103 Defined as: this node is an ancestor of or identical to the other node.
104 """
105 if not isinstance(other, Graphable):
106 return NotImplemented
108 return self is other or self.find_path(other) is not None
110 def __gt__(self, other: object) -> bool:
111 """
112 Check if this node is 'greater than' another.
113 Defined as: this node is a proper descendant of the other node.
114 """
115 if not isinstance(other, Graphable):
116 return NotImplemented
118 return other.find_path(self) is not None
120 def __ge__(self, other: object) -> bool:
121 """
122 Check if this node is 'greater than or equal' to another.
123 Defined as: this node is a descendant of or identical to the other node.
124 """
125 if not isinstance(other, Graphable):
126 return NotImplemented
128 return self is other or other.find_path(self) is not None
130 def add_dependencies(
131 self, dependencies: set[Self], check_cycles: bool = False, **attributes: Any
132 ) -> None:
133 """
134 Add multiple dependencies to this node.
136 Args:
137 dependencies (set[Self]): A set of Graphable nodes to add as dependencies.
138 check_cycles (bool): If True, check if adding these dependencies would create a cycle.
139 **attributes: Edge attributes to apply to all added dependencies.
140 """
141 logger.debug(f"Node '{self.reference}': adding dependencies {dependencies}")
142 for dependency in dependencies:
143 self.add_dependency(dependency, check_cycles=check_cycles, **attributes)
145 def add_dependency(
146 self, dependency: Self, check_cycles: bool = False, **attributes: Any
147 ) -> None:
148 """
149 Add a single dependency to this node.
151 Args:
152 dependency (Self): The Graphable node to add as a dependency.
153 check_cycles (bool): If True, check if adding this dependency would create a cycle.
154 **attributes: Edge attributes (e.g., weight, label).
156 Raises:
157 GraphCycleError: If check_cycles is True and a cycle would be created.
158 """
159 if check_cycles:
160 logger.debug(
161 f"Checking if adding dependency '{dependency.reference}' to '{self.reference}' creates a cycle"
162 )
163 if path := self.find_path(dependency):
164 cycle = path + [self]
165 logger.error(
166 f"Cycle detected: {' -> '.join(str(n.reference) for n in cycle)}"
167 )
168 raise GraphCycleError(
169 f"Adding dependency '{dependency.reference}' to "
170 f"'{self.reference}' would create a cycle.",
171 cycle=cycle,
172 )
174 logger.debug(
175 f"Node '{self.reference}': adding dependency '{dependency.reference}' with attributes {attributes}"
176 )
177 self._add_depends_on(dependency, **attributes)
178 dependency._add_dependent(self, **attributes)
180 def add_dependent(
181 self, dependent: Self, check_cycles: bool = False, **attributes: Any
182 ) -> None:
183 """
184 Add a single dependent to this node.
186 Args:
187 dependent (Self): The Graphable node to add as a dependent.
188 check_cycles (bool): If True, check if adding this dependent would create a cycle.
189 **attributes: Edge attributes (e.g., weight, label).
191 Raises:
192 GraphCycleError: If check_cycles is True and a cycle would be created.
193 """
194 if check_cycles:
195 logger.debug(
196 f"Checking if adding dependent '{dependent.reference}' to '{self.reference}' creates a cycle"
197 )
198 if path := dependent.find_path(self):
199 cycle = path + [dependent]
200 logger.error(
201 f"Cycle detected: {' -> '.join(str(n.reference) for n in cycle)}"
202 )
203 raise GraphCycleError(
204 f"Adding dependent '{dependent.reference}' to "
205 f"'{self.reference}' would create a cycle.",
206 cycle=cycle,
207 )
209 logger.debug(
210 f"Node '{self.reference}': adding dependent '{dependent.reference}' with attributes {attributes}"
211 )
212 self._add_dependent(dependent, **attributes)
213 dependent._add_depends_on(self, **attributes)
215 def add_dependents(
216 self, dependents: set[Self], check_cycles: bool = False, **attributes: Any
217 ) -> None:
218 """
219 Add multiple dependents to this node.
221 Args:
222 dependents (set[Self]): A set of Graphable nodes to add as dependents.
223 check_cycles (bool): If True, check if adding these dependents would create a cycle.
224 **attributes: Edge attributes to apply to all added dependents.
225 """
226 logger.debug(f"Node '{self.reference}': adding dependents {dependents}")
227 for dependent in dependents:
228 self.add_dependent(dependent, check_cycles=check_cycles, **attributes)
230 def _add_dependent(self, dependent: Self, **attributes: Any) -> None:
231 """
232 Internal method to add a dependent node (incoming edge in dependency graph).
234 Args:
235 dependent (Self): The node that depends on this node.
236 **attributes: Edge attributes.
237 """
238 if (
239 dependent not in self._dependents
240 or self._dependents[dependent] != attributes
241 ):
242 self._dependents[dependent] = attributes
243 logger.debug(
244 f"Node '{self.reference}': added dependent '{dependent.reference}' with attributes {attributes}"
245 )
246 self._notify_change()
248 def _add_depends_on(self, depends_on: Self, **attributes: Any) -> None:
249 """
250 Internal method to add a dependency (outgoing edge in dependency graph).
252 Args:
253 depends_on (Self): The node that this node depends on.
254 **attributes: Edge attributes.
255 """
256 if (
257 depends_on not in self._depends_on
258 or self._depends_on[depends_on] != attributes
259 ):
260 self._depends_on[depends_on] = attributes
261 logger.debug(
262 f"Node '{self.reference}': added dependency '{depends_on.reference}' with attributes {attributes}"
263 )
264 self._notify_change()
266 def _remove_dependent(self, dependent: Self) -> None:
267 """
268 Internal method to remove a dependent node.
270 Args:
271 dependent (Self): The node to remove.
272 """
273 if dependent in self._dependents:
274 del self._dependents[dependent]
275 logger.debug(
276 f"Node '{self.reference}': removed dependent '{dependent.reference}'"
277 )
278 self._notify_change()
280 def _remove_depends_on(self, depends_on: Self) -> None:
281 """
282 Internal method to remove a dependency.
284 Args:
285 depends_on (Self): The node to remove.
286 """
287 if depends_on in self._depends_on:
288 del self._depends_on[depends_on]
289 logger.debug(
290 f"Node '{self.reference}': removed dependency '{depends_on.reference}'"
291 )
292 self._notify_change()
294 def _register_observer(self, observer: GraphObserver) -> None:
295 """Register an observer to be notified of changes."""
296 self._observers.add(observer)
298 def _unregister_observer(self, observer: GraphObserver) -> None:
299 """Unregister an observer."""
300 self._observers.discard(observer)
302 def add_tag(self, tag: str) -> None:
303 """
304 Add a tag to this node.
306 Args:
307 tag (str): The tag to add.
308 """
309 self._tags.add(tag)
310 logger.debug(f"Added tag '{tag}' to {self.reference}")
311 self._notify_change()
313 @property
314 def dependents(self) -> set[Self]:
315 """
316 Get the set of nodes that depend on this node.
318 Returns:
319 set[Self]: A copy of the dependents set.
320 """
321 return cast(set[Self], set(self._dependents))
323 @property
324 def depends_on(self) -> set[Self]:
325 """
326 Get the set of nodes that this node depends on.
328 Returns:
329 set[Self]: A copy of the dependencies set.
330 """
331 return cast(set[Self], set(self._depends_on))
333 def is_tagged(self, tag: str) -> bool:
334 """
335 Check if the node has a specific tag.
337 Args:
338 tag (str): The tag to check.
340 Returns:
341 bool: True if the tag exists, False otherwise.
342 """
343 logger.debug(f"Node '{self.reference}': checking if tagged with '{tag}'")
344 return tag in self._tags
346 def edge_attributes(self, other: Self) -> dict[str, Any]:
347 """
348 Get the attributes of the edge between this node and another.
349 Checks both outgoing (dependents) and incoming (depends_on) edges.
351 Args:
352 other (Self): The other node.
354 Returns:
355 dict[str, Any]: The edge attributes.
357 Raises:
358 KeyError: If no edge exists between the nodes.
359 """
360 if other in self._dependents:
361 return self._dependents[other]
362 if other in self._depends_on:
363 return self._depends_on[other]
364 raise KeyError(f"No edge between '{self.reference}' and '{other.reference}'")
366 def set_edge_attribute(self, other: Self, key: str, value: Any) -> None:
367 """
368 Set an attribute on the edge between this node and another.
370 Args:
371 other (Self): The other node.
372 key (str): The attribute key.
373 value (Any): The attribute value.
374 """
375 if other in self._dependents:
376 self._dependents[other][key] = value
377 other._depends_on[self][key] = value
378 self._notify_change()
379 elif other in self._depends_on:
380 self._depends_on[other][key] = value
381 other._dependents[self][key] = value
382 self._notify_change()
383 else:
384 raise KeyError(
385 f"No edge between '{self.reference}' and '{other.reference}'"
386 )
388 def find_path(self, target: Self) -> list[Self] | None:
389 """
390 Find a path from this node to the target node using BFS.
392 Args:
393 target (Self): The target node to find a path to.
395 Returns:
396 list[Self] | None: The shortest path as a list of nodes, or None if no path exists.
397 """
398 logger.debug(
399 f"Searching for path from '{self.reference}' to '{target.reference}'"
400 )
401 queue: deque[tuple[Self, list[Self]]] = deque()
402 for neighbor in self._dependents:
403 # neighbor is Graphable[Any], Self is a bound of Graphable[T]
404 queue.append((cast(Self, neighbor), [self, cast(Self, neighbor)]))
406 visited: set[Graphable[Any]] = set()
407 while queue:
408 current, path = queue.popleft()
409 if current == target:
410 logger.debug(
411 f"Path found: {' -> '.join(str(n.reference) for n in path)}"
412 )
413 return path
415 if current in visited:
416 continue
418 visited.add(current)
419 for neighbor in current.dependents:
420 if neighbor not in visited:
421 queue.append((neighbor, path + [neighbor]))
423 logger.debug(f"No path found from '{self.reference}' to '{target.reference}'")
424 return None
426 def provides_to(self, dependent: Self, check_cycles: bool = False) -> None:
427 """
428 Alias for add_dependent: indicates this node provides something to another.
430 Args:
431 dependent (Self): The Graphable node that receives the provision.
432 check_cycles (bool): If True, check if adding this dependent would create a cycle.
433 """
434 logger.debug(
435 f"Node '{self.reference}': providing to '{dependent.reference}' (via alias)"
436 )
437 self.add_dependent(dependent, check_cycles=check_cycles)
439 @property
440 def reference(self) -> T:
441 """
442 Get the underlying reference object.
444 Returns:
445 T: The reference object.
446 """
447 return self._reference
449 def requires(self, dependency: Self, check_cycles: bool = False) -> None:
450 """
451 Alias for add_dependency: indicates this node requires another node.
453 Args:
454 dependency (Self): The Graphable node that is required.
455 check_cycles (bool): If True, check if adding this dependency would create a cycle.
456 """
457 logger.debug(
458 f"Node '{self.reference}': requiring dependency '{dependency.reference}' (via alias)"
459 )
460 self.add_dependency(dependency, check_cycles=check_cycles)
462 @property
463 def tags(self) -> set[str]:
464 """
465 Get the set of tags for this node.
467 Returns:
468 set[str]: A copy of the tags set.
469 """
470 return set(self._tags)
472 def remove_tag(self, tag: str) -> None:
473 """
474 Remove a tag from this node.
476 Args:
477 tag (str): The tag to remove.
478 """
479 if tag in self._tags:
480 self._tags.discard(tag)
481 logger.debug(f"Removed tag '{tag}' from {self.reference}")
482 self._notify_change()