Coverage for collectra/types/links.py: 80.88%

1from __future__ import annotations 

2 

3from dataclasses import dataclass, field 

4from pathlib import Path 

5from typing import Any 

6 

7from .base import Artefact 

8 

9__all__ = ["Link"] 

10 

11 

12@dataclass 

13class Link(Artefact): 

14 """A soft reference to the artefact identified by its first parent.""" 

15 

16 target: Artefact | None = field(default=None, repr=False, compare=False) 

17 

18 @property 

19 def parent_id(self) -> str: 

20 if isinstance(self.parents, str): 

21 return self.parents 

22 return self.parents[0] if self.parents else "" 

23 

24 @property 

25 def __class__(self) -> type: 

26 """Present the resolved target type to ``isinstance`` consumers.""" 

27 target = self.__dict__.get("target") 

28 seen = {id(self)} 

29 while type(target) is Link: 

30 if id(target) in seen: 

31 return Link 

32 seen.add(id(target)) 

33 target = target.__dict__.get("target") 

34 return type(target) if target is not None else Link 

35 

36 def bind(self, artefacts: dict[str, Artefact]) -> Artefact | None: 

37 """Resolve the first parent ID and retain it as the runtime target.""" 

38 self.target = artefacts.get(self.parent_id) 

39 return self.target 

40 

41 def resolve(self) -> Artefact: 

42 """Follow any number of Links and return the concrete artefact.""" 

43 current: Artefact = self 

44 chain: list[str] = [] 

45 seen: set[int] = set() 

46 while type(current) is Link: 

47 if id(current) in seen: 

48 chain.append(current.id) 

49 raise RuntimeError(f"Cyclic Link chain: {' -> '.join(chain)}") 

50 seen.add(id(current)) 

51 chain.append(current.id) 

52 target = current.__dict__.get("target") 

53 if target is None: 

54 raise RuntimeError( 

55 f"Link chain {' -> '.join(chain)!r} is unresolved at " 

56 f"parent {current.parent_id!r}" 

57 ) 

58 current = target 

59 return current 

60 

61 def __call__(self) -> Any: 

62 return self.resolve()() 

63 

64 def display(self, context) -> dict: 

65 return self.resolve().display(context) 

66 

67 def __getattr__(self, name: str) -> Any: 

68 # Dataclass initialization probes attributes before ``target`` exists. 

69 target = self.__dict__.get("target") 

70 if target is None: 

71 raise AttributeError(name) 

72 return getattr(self.resolve(), name) 

73 

74 def attributes_to_ignore(self) -> set: 

75 attributes = super().attributes_to_ignore() 

76 attributes.add("target") 

77 return attributes 

78 

79 def serialize(self) -> dict: 

80 """Serialize only the soft-link identity and parent reference.""" 

81 serialized = { 

82 "type": self.get_class_path(), 

83 "id": self.id, 

84 } 

85 if self.parents: 

86 serialized["parents"] = ( 

87 self.parents 

88 if isinstance(self.parents, str) 

89 else self.parents[0] if len(self.parents) == 1 else self.parents 

90 ) 

91 return serialized 

92 

93 def extract(self, path: Path | str) -> None: 

94 self.resolve().extract(path) 

95 

96 def evaluate(self, gold: Artefact) -> float: 

97 if type(gold) is Link: 

98 if self.target is not None and gold.target is not None: 

99 return self.resolve().evaluate(gold.resolve()) 

100 return float(bool(self.parent_id) and self.parent_id == gold.parent_id) 

101 return self.resolve().evaluate(gold)