Coverage for tarjinja/iface.py: 92%

1import copy 

2import fnmatch 

3from collections.abc import Generator 

4from logging import getLogger 

5 

6import braceexpand 

7 

8log = getLogger(__name__) 

9 

10 

11class Input: 

12 def __init__(self, ifn: str, **kwargs): 

13 self.ifn = ifn 

14 

15 def walk(self) -> Generator[tuple[str, int, float], None, None]: 

16 raise NotImplementedError("walk") 

17 

18 def readfile(self, fn: str) -> str: 

19 raise NotImplementedError("readfile") 

20 

21 

22class Filter: 

23 def __init__(self, **kwargs): 

24 pass 

25 

26 def render(self, s: str, vals: dict) -> str: 

27 raise NotImplementedError("render") 

28 

29 def renderfn(self, s: str, vals: dict) -> Generator[str, None, None]: 

30 return braceexpand.braceexpand(self.render(s, vals)) 

31 

32 def strtr(self, strng: str, replace: dict) -> str: 

33 # https://stackoverflow.com/questions/10931150/phps-strtr-for-python 

34 buffer = [] 

35 i, n = 0, len(strng) 

36 while i < n: 

37 match = False 

38 for s, r in replace.items(): 

39 if strng[i : len(s) + i] == s: 

40 buffer.append(r) 

41 i = i + len(s) 

42 match = True 

43 break 

44 if not match: 

45 buffer.append(strng[i]) 

46 i = i + 1 

47 return "".join(buffer) 

48 

49 

50class Output: 

51 def __init__(self, ofn: str, **kwargs): 

52 self.ofn = ofn 

53 

54 def writefile(self, fn: str, content: str, mode: int, ts: float | None = None): 

55 raise NotImplementedError("writefile") 

56 

57 def finish(self): 

58 log.debug("finished %s", self.ofn) 

59 

60 

61class Pipeline: 

62 def __init__( 

63 self, inp: Input, filt: Filter, outp: Output, passpat: str | None = None 

64 ): 

65 self.inp = inp 

66 self.filt = filt 

67 self.outp = outp 

68 self.passpat = passpat 

69 

70 def render(self, vals: dict): 

71 for fnpat, mode, ts in self.inp.walk(): 

72 log.debug("walk %s (%o)", fnpat, mode) 

73 content = self.inp.readfile(fnpat) 

74 for fn in self.filt.renderfn(fnpat, vals): 

75 if self.passpat is not None and fnmatch.fnmatch(fn, self.passpat): 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 self.outp.writefile(fnpat, content, mode, ts) 

77 else: 

78 v = copy.deepcopy(vals) 

79 v["fname"] = fn 

80 ocont = self.filt.render(content, v) 

81 log.debug("write %s", fn) 

82 self.outp.writefile(fn, ocont, mode, ts) 

83 self.outp.finish()