Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1""" 

2Tracing utils 

3""" 

4from .callers import _Result 

5 

6 

7class TagTracer(object): 

8 def __init__(self): 

9 self._tag2proc = {} 

10 self.writer = None 

11 self.indent = 0 

12 

13 def get(self, name): 

14 return TagTracerSub(self, (name,)) 

15 

16 def format_message(self, tags, args): 

17 if isinstance(args[-1], dict): 

18 extra = args[-1] 

19 args = args[:-1] 

20 else: 

21 extra = {} 

22 

23 content = " ".join(map(str, args)) 

24 indent = " " * self.indent 

25 

26 lines = ["%s%s [%s]\n" % (indent, content, ":".join(tags))] 

27 

28 for name, value in extra.items(): 

29 lines.append("%s %s: %s\n" % (indent, name, value)) 

30 return lines 

31 

32 def processmessage(self, tags, args): 

33 if self.writer is not None and args: 

34 lines = self.format_message(tags, args) 

35 self.writer("".join(lines)) 

36 try: 

37 self._tag2proc[tags](tags, args) 

38 except KeyError: 

39 pass 

40 

41 def setwriter(self, writer): 

42 self.writer = writer 

43 

44 def setprocessor(self, tags, processor): 

45 if isinstance(tags, str): 

46 tags = tuple(tags.split(":")) 

47 else: 

48 assert isinstance(tags, tuple) 

49 self._tag2proc[tags] = processor 

50 

51 

52class TagTracerSub(object): 

53 def __init__(self, root, tags): 

54 self.root = root 

55 self.tags = tags 

56 

57 def __call__(self, *args): 

58 self.root.processmessage(self.tags, args) 

59 

60 def setmyprocessor(self, processor): 

61 self.root.setprocessor(self.tags, processor) 

62 

63 def get(self, name): 

64 return self.__class__(self.root, self.tags + (name,)) 

65 

66 

67class _TracedHookExecution(object): 

68 def __init__(self, pluginmanager, before, after): 

69 self.pluginmanager = pluginmanager 

70 self.before = before 

71 self.after = after 

72 self.oldcall = pluginmanager._inner_hookexec 

73 assert not isinstance(self.oldcall, _TracedHookExecution) 

74 self.pluginmanager._inner_hookexec = self 

75 

76 def __call__(self, hook, hook_impls, kwargs): 

77 self.before(hook.name, hook_impls, kwargs) 

78 outcome = _Result.from_call(lambda: self.oldcall(hook, hook_impls, kwargs)) 

79 self.after(outcome, hook.name, hook_impls, kwargs) 

80 return outcome.get_result() 

81 

82 def undo(self): 

83 self.pluginmanager._inner_hookexec = self.oldcall