Coverage for graphqler / chains / chain.py: 71%

1"""Chain dataclass representing an ordered sequence of nodes to execute.""" 

2 

3from dataclasses import dataclass, field 

4from graphqler.graph.node import Node 

5 

6 

7@dataclass 

8class Chain: 

9 """An ordered list of nodes representing a root-to-leaf path (or a prefix thereof). 

10 

11 Example: for a dependency path A -> B -> C, the following chains would be generated: 

12 Chain([A]), Chain([A, B]), Chain([A, B, C]) 

13 

14 Each chain is intended to be executed from start to finish with a fresh ObjectsBucket, 

15 ensuring all prerequisite nodes run before dependent ones within the same chain. 

16 """ 

17 

18 nodes: list[Node] = field(default_factory=list) 

19 name: str = "" 

20 

21 def __len__(self) -> int: 

22 return len(self.nodes) 

23 

24 def __repr__(self) -> str: 

25 path = " -> ".join(n.name for n in self.nodes) 

26 return f"Chain([{path}])" 

27 

28 def last_node(self) -> Node | None: 

29 """Returns the terminal (last) node in the chain, or None if the chain is empty.""" 

30 if not self.nodes: 

31 return None 

32 return self.nodes[-1] 

33 

34 def has_mutation_type(self, mutation_types: list[str]) -> bool: 

35 """Returns True if any node in the chain has a mutation_type in the given list. 

36 

37 Args: 

38 mutation_types (list[str]): Mutation types to check for (e.g. ["UPDATE", "DELETE"]) 

39 

40 Returns: 

41 bool: True if any node matches 

42 """ 

43 return any(n.mutation_type in mutation_types for n in self.nodes)