Coverage for github_test/pipeline.py: 100%

1"""An example module: a tiny text-transform pipeline. 

2 

3This is throwaway "replace-me" starter code. It exists to show how the project is 

4wired together -- in particular how Hydra's ``instantiate`` builds a *nested* object 

5graph from config: a ``Pipeline`` whose ``steps`` are themselves ``_target_`` objects 

6(see ``configs/pipeline.yaml``). Delete it and drop in your own package code. 

7""" 

8 

9from abc import ABC, abstractmethod 

10from dataclasses import dataclass 

11 

12from loguru import logger 

13 

14 

15class Transform(ABC): 

16 """A single pipeline step that maps a string to a string.""" 

17 

18 @abstractmethod 

19 def __call__(self, text: str) -> str: 

20 """Apply the transform to ``text`` and return the result.""" 

21 

22 

23class Upper(Transform): 

24 """Upper-cases the text.""" 

25 

26 def __call__(self, text: str) -> str: 

27 """Return ``text`` upper-cased.""" 

28 return text.upper() 

29 

30 

31class Repeat(Transform): 

32 """Repeats the text ``times`` times, space-separated.""" 

33 

34 def __init__(self, times: int = 2) -> None: 

35 """Store how many times to repeat the text.""" 

36 self.times = times 

37 

38 def __call__(self, text: str) -> str: 

39 """Return ``text`` repeated ``times`` times, space-separated.""" 

40 return " ".join([text] * self.times) 

41 

42 

43@dataclass 

44class Pipeline: 

45 """Runs an ordered list of transforms over a starting message.""" 

46 

47 steps: list[Transform] 

48 message: str = "hello" 

49 

50 def run(self) -> str: 

51 """Apply each transform in turn, log the result, and return it.""" 

52 result = self.message 

53 for step in self.steps: 

54 result = step(result) 

55 logger.info(result) 

56 return result