##// END OF EJS Templates
ZMQHistoryManager made to work with testing KernelClient which has no history method
Doug Blank -
Show More
@@ -1,93 +1,95
1 1 """ ZMQ Kernel History accessor and manager. """
2 2 #-----------------------------------------------------------------------------
3 3 # Copyright (C) 2010-2011 The IPython Development Team.
4 4 #
5 5 # Distributed under the terms of the BSD License.
6 6 #
7 7 # The full license is in the file COPYING.txt, distributed with this software.
8 8 #-----------------------------------------------------------------------------
9 9
10 10 #-----------------------------------------------------------------------------
11 11 # Imports
12 12 #-----------------------------------------------------------------------------
13 13
14 14 from IPython.core.history import HistoryAccessorBase
15 15 from IPython.utils.traitlets import Dict, List
16 16
17 17 try:
18 18 from queue import Empty # Py 3
19 19 except ImportError:
20 20 from Queue import Empty # Py 2
21 21
22 22 class ZMQHistoryManager(HistoryAccessorBase):
23 23 """History accessor and manager for ZMQ-based kernels"""
24 24 input_hist_parsed = List([""])
25 25 output_hist = Dict()
26 26 dir_hist = List()
27 27 output_hist_reprs = Dict()
28 28
29 29 def __init__(self, client):
30 30 """
31 31 Class to load the command-line history from a ZMQ-based kernel,
32 32 and access the history.
33 33
34 34 Parameters
35 35 ----------
36 36
37 37 client: `IPython.kernel.KernelClient`
38 38 The kernel client in order to request the history.
39 39 """
40 40 self.client = client
41 41
42 42 def _load_history(self, raw=True, output=False, hist_access_type='range',
43 43 **kwargs):
44 44 """
45 45 Load the history over ZMQ from the kernel. Wraps the history
46 46 messaging with loop to wait to get history results.
47 47 """
48 history = []
49 if hasattr(self.client, "history"):
50 ## In tests, KernelClient may not have a history method
48 51 msg_id = self.client.history(raw=raw, output=output,
49 52 hist_access_type=hist_access_type,
50 53 **kwargs)
51 history = []
52 54 while True:
53 55 try:
54 56 reply = self.client.get_shell_msg(timeout=1)
55 57 except Empty:
56 58 break
57 59 else:
58 60 if reply['parent_header'].get('msg_id') == msg_id:
59 61 history = reply['content'].get('history', [])
60 62 break
61 63 return history
62 64
63 65 def get_tail(self, n=10, raw=True, output=False, include_latest=False):
64 66 return self._load_history(hist_access_type='tail', n=n, raw=raw,
65 67 output=output)
66 68
67 69 def search(self, pattern="*", raw=True, search_raw=True,
68 70 output=False, n=None, unique=False):
69 71 return self._load_history(hist_access_type='search', pattern=pattern,
70 72 raw=raw, search_raw=search_raw,
71 73 output=output, n=n, unique=unique)
72 74
73 75 def get_range(self, session, start=1, stop=None, raw=True,output=False):
74 76 return self._load_history(hist_access_type='range', raw=raw,
75 77 output=output, start=start, stop=stop,
76 78 session=session)
77 79
78 80 def get_range_by_str(self, rangestr, raw=True, output=False):
79 81 return self._load_history(hist_access_type='range', raw=raw,
80 82 output=output, rangestr=rangestr)
81 83
82 84 def end_session(self):
83 85 """
84 86 Nothing to do for ZMQ-based histories.
85 87 """
86 88 pass
87 89
88 90 def reset(self, new_session=True):
89 91 """
90 92 Nothing to do for ZMQ-based histories.
91 93 """
92 94 pass
93 95
General Comments 0
You need to be logged in to leave comments. Login now