Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/IPython/core/history.py: 31%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""History related magics and functionality"""
3from __future__ import annotations
5# Copyright (c) IPython Development Team.
6# Distributed under the terms of the Modified BSD License.
9import atexit
10import datetime
11import os
12import re
13import weakref
16import threading
17from pathlib import Path
19import functools
20from collections import defaultdict
21from contextlib import contextmanager
22from dataclasses import dataclass
23from traitlets import (
24 Any,
25 Bool,
26 Dict,
27 Instance,
28 Integer,
29 List,
30 TraitError,
31 Unicode,
32 Union,
33 default,
34 observe,
35)
36from traitlets.config.configurable import LoggingConfigurable
38from IPython.paths import locate_profile
39from IPython.utils.decorators import undoc
40from typing import TYPE_CHECKING, ParamSpec
41from collections.abc import Iterable
42import typing
43import typing as t
44from typing import cast
45from warnings import warn
46from weakref import ref, WeakSet
48if TYPE_CHECKING:
49 from types import TracebackType
51 from IPython.core.interactiveshell import InteractiveShell
52 from traitlets.config import Config as Configuration
54try:
55 from sqlite3 import DatabaseError, OperationalError
56 import sqlite3
58 sqlite3.register_converter(
59 "timestamp", lambda val: datetime.datetime.fromisoformat(val.decode())
60 )
62 sqlite3_found = True
63except ModuleNotFoundError:
64 sqlite3_found = False
66 class DatabaseError(Exception): # type: ignore [no-redef]
67 pass
69 class OperationalError(Exception): # type: ignore [no-redef]
70 pass
73InOrInOut = str | tuple[str, str | None]
75# -----------------------------------------------------------------------------
76# Classes and functions
77# -----------------------------------------------------------------------------
80@undoc
81class DummyDB:
82 """Dummy DB that will act as a black hole for history.
84 Only used in the absence of sqlite"""
86 def execute(*args: typing.Any, **kwargs: typing.Any) -> list:
87 return []
89 def commit(self, *args: typing.Any, **kwargs: typing.Any) -> None:
90 pass
92 def __enter__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
93 pass
95 def __exit__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
96 pass
98 def close(self, *args: typing.Any, **kwargs: typing.Any) -> None:
99 pass
102_P = ParamSpec("_P")
103_R = t.TypeVar("_R")
106def only_when_enabled(f: t.Callable[_P, _R]) -> t.Callable[_P, _R]:
107 """Decorator: return an empty list in the absence of sqlite.
109 Typed as signature-preserving (like the ``decorator``-package version it
110 replaces): the empty-list fallback for a disabled accessor is invisible to
111 the type system, as before.
112 """
114 @functools.wraps(f)
115 def wrapper(*a: _P.args, **kw: _P.kwargs) -> _R:
116 self = cast("HistoryAccessor", a[0])
117 if not self.enabled:
118 return cast(_R, [])
119 else:
120 return f(*a, **kw)
122 return wrapper
125# use 16kB as threshold for whether a corrupt history db should be saved
126# that should be at least 100 entries or so
127_SAVE_DB_SIZE = 16384
130def catch_corrupt_db(f: t.Callable[_P, _R]) -> t.Callable[_P, _R]:
131 """A decorator which wraps HistoryAccessor method calls to catch errors from
132 a corrupt SQLite database, move the old database out of the way, and create
133 a new one.
135 We avoid clobbering larger databases because this may be triggered due to filesystem issues,
136 not just a corrupt file.
137 """
139 @functools.wraps(f)
140 def wrapper(*a: _P.args, **kw: _P.kwargs) -> _R:
141 self = cast("HistoryAccessor", a[0])
142 try:
143 return f(*a, **kw)
144 except (DatabaseError, OperationalError) as e:
145 self._corrupt_db_counter += 1
146 self.log.error("Failed to open SQLite history %s (%s).", self.hist_file, e)
147 if self.hist_file != ":memory:":
148 if self._corrupt_db_counter > self._corrupt_db_limit:
149 self.hist_file = ":memory:"
150 self.log.error(
151 "Failed to load history too many times, history will not be saved."
152 )
153 elif self.hist_file.is_file():
154 # move the file out of the way
155 base = str(self.hist_file.parent / self.hist_file.stem)
156 ext = self.hist_file.suffix
157 size = self.hist_file.stat().st_size
158 if size >= _SAVE_DB_SIZE:
159 # if there's significant content, avoid clobbering
160 now = (
161 datetime.datetime.now(datetime.UTC)
162 .isoformat()
163 .replace(":", ".")
164 )
165 newpath = base + "-corrupt-" + now + ext
166 # don't clobber previous corrupt backups
167 for i in range(100):
168 if not Path(newpath).exists():
169 break
170 else:
171 newpath = base + "-corrupt-" + now + ("-%i" % i) + ext
172 else:
173 # not much content, possibly empty; don't worry about clobbering
174 # maybe we should just delete it?
175 newpath = base + "-corrupt" + ext
176 self.hist_file.rename(newpath)
177 self.log.error(
178 "History file was moved to %s and a new file created.", newpath
179 )
180 self.init_db()
181 return cast(_R, [])
182 else:
183 # Failed with :memory:, something serious is wrong
184 raise
186 return wrapper
189class HistoryAccessorBase(LoggingConfigurable):
190 """An abstract class for History Accessors"""
192 def get_tail(
193 self,
194 n: int = 10,
195 raw: bool = True,
196 output: bool = False,
197 include_latest: bool = False,
198 ) -> Iterable[tuple[int, int, InOrInOut]]:
199 raise NotImplementedError
201 def search(
202 self,
203 pattern: str = "*",
204 raw: bool = True,
205 search_raw: bool = True,
206 output: bool = False,
207 n: int | None = None,
208 unique: bool = False,
209 ) -> Iterable[tuple[int, int, InOrInOut]]:
210 raise NotImplementedError
212 def get_range(
213 self,
214 session: int,
215 start: int = 1,
216 stop: int | None = None,
217 raw: bool = True,
218 output: bool = False,
219 ) -> Iterable[tuple[int, int, InOrInOut]]:
220 raise NotImplementedError
222 def get_range_by_str(
223 self, rangestr: str, raw: bool = True, output: bool = False
224 ) -> Iterable[tuple[int, int, InOrInOut]]:
225 raise NotImplementedError
228class HistoryAccessor(HistoryAccessorBase):
229 """Access the history database without adding to it.
231 This is intended for use by standalone history tools. IPython shells use
232 HistoryManager, below, which is a subclass of this."""
234 # counter for init_db retries, so we don't keep trying over and over
235 _corrupt_db_counter = 0
236 # after two failures, fallback on :memory:
237 _corrupt_db_limit = 2
239 # String holding the path to the history file
240 hist_file = Union(
241 [Instance(Path), Unicode()],
242 help="""Path to file to use for SQLite history database.
244 By default, IPython will put the history database in the IPython
245 profile directory. If you would rather share one history among
246 profiles, you can set this value in each, so that they are consistent.
248 Due to an issue with fcntl, SQLite is known to misbehave on some NFS
249 mounts. If you see IPython hanging, try setting this to something on a
250 local disk, e.g::
252 ipython --HistoryManager.hist_file=/tmp/ipython_hist.sqlite
254 you can also use the specific value `:memory:` (including the colon
255 at both end but not the back ticks), to avoid creating an history file.
257 """,
258 ).tag(config=True)
260 enabled = Bool(
261 sqlite3_found,
262 help="""enable the SQLite history
264 set enabled=False to disable the SQLite history,
265 in which case there will be no stored history, no SQLite connection,
266 and no background saving thread. This may be necessary in some
267 threaded environments where IPython is embedded.
268 """,
269 ).tag(config=True)
271 connection_options = Dict(
272 help="""Options for configuring the SQLite connection
274 These options are passed as keyword args to sqlite3.connect
275 when establishing database connections.
276 """
277 ).tag(config=True)
279 @default("connection_options")
280 def _default_connection_options(self) -> dict[str, bool]:
281 return dict(check_same_thread=False)
283 # The SQLite database
284 db = Any()
286 @observe("db")
287 @only_when_enabled
288 def _db_changed(self, change): # type: ignore [no-untyped-def]
289 """validate the db, since it can be an Instance of two different types"""
290 new = change["new"]
291 connection_types = (DummyDB, sqlite3.Connection)
292 if not isinstance(new, connection_types):
293 msg = "{}.db must be sqlite3 Connection or DummyDB, not {!r}".format(
294 self.__class__.__name__,
295 new,
296 )
297 raise TraitError(msg)
299 def __init__(
300 self, profile: str = "default", hist_file: str = "", **traits: typing.Any
301 ) -> None:
302 """Create a new history accessor.
304 Parameters
305 ----------
306 profile : str
307 The name of the profile from which to open history.
308 hist_file : str
309 Path to an SQLite history database stored by IPython. If specified,
310 hist_file overrides profile.
311 config : :class:`~traitlets.config.loader.Config`
312 Config object. hist_file can also be set through this.
313 """
314 super().__init__(**traits)
315 # defer setting hist_file from kwarg until after init,
316 # otherwise the default kwarg value would clobber any value
317 # set by config
318 if hist_file:
319 self.hist_file = hist_file
321 try:
322 self.hist_file
323 except TraitError:
324 # No one has set the hist_file, yet.
325 self.hist_file = self._get_hist_file_name(profile)
327 self.init_db()
329 def _get_hist_file_name(self, profile: str = "default") -> Path:
330 """Find the history file for the given profile name.
332 This is overridden by the HistoryManager subclass, to use the shell's
333 active profile.
335 Parameters
336 ----------
337 profile : str
338 The name of a profile which has a history file.
339 """
340 return Path(locate_profile(profile)) / "history.sqlite"
342 @catch_corrupt_db
343 def init_db(self) -> None:
344 """Connect to the database, and create tables if necessary."""
345 if not self.enabled:
346 self.db = DummyDB()
347 self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db)
348 return
350 # use detect_types so that timestamps return datetime objects
351 kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES)
352 kwargs.update(self.connection_options)
353 self.db = sqlite3.connect(str(self.hist_file), **kwargs) # type: ignore [call-overload]
354 self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db)
355 with self.db:
356 self.db.execute(
357 """CREATE TABLE IF NOT EXISTS sessions (session integer
358 primary key autoincrement, start timestamp,
359 end timestamp, num_cmds integer, remark text)"""
360 )
361 self.db.execute(
362 """CREATE TABLE IF NOT EXISTS history
363 (session integer, line integer, source text, source_raw text,
364 PRIMARY KEY (session, line))"""
365 )
366 # Output history is optional, but ensure the table's there so it can be
367 # enabled later.
368 self.db.execute(
369 """CREATE TABLE IF NOT EXISTS output_history
370 (session integer, line integer, output text,
371 PRIMARY KEY (session, line))"""
372 )
373 # success! reset corrupt db count
374 self._corrupt_db_counter = 0
376 def close(self) -> None:
377 """Close the SQLite database connection.
379 Prefer calling this to closing ``self.db`` directly: it gives
380 subclasses (notably :class:`HistoryManager`) a single place to hook in
381 the rest of their teardown. Safe to call more than once.
382 """
383 self.db.close()
385 def __enter__(self) -> HistoryAccessor:
386 """Support use as a context manager for deterministic cleanup::
388 with HistoryAccessor(hist_file=path) as history:
389 ...
390 # connection is closed here
392 :class:`HistoryManager` additionally stops its saving thread on exit.
393 """
394 return self
396 def __exit__(
397 self,
398 exc_type: type[BaseException] | None,
399 exc_value: BaseException | None,
400 traceback: TracebackType | None,
401 ) -> None:
402 self.close()
404 def writeout_cache(self) -> None:
405 """Overridden by HistoryManager to dump the cache before certain
406 database lookups."""
407 pass
409 ## -------------------------------
410 ## Methods for retrieving history:
411 ## -------------------------------
412 def _run_sql(
413 self,
414 sql: str,
415 params: tuple,
416 raw: bool = True,
417 output: bool = False,
418 latest: bool = False,
419 ) -> Iterable[tuple[int, int, InOrInOut]]:
420 """Prepares and runs an SQL query for the history database.
422 Parameters
423 ----------
424 sql : str
425 Any filtering expressions to go after SELECT ... FROM ...
426 params : tuple
427 Parameters passed to the SQL query (to replace "?")
428 raw, output : bool
429 See :meth:`get_range`
430 latest : bool
431 Select rows with max (session, line)
433 Returns
434 -------
435 Tuples as :meth:`get_range`
436 """
437 toget = "source_raw" if raw else "source"
438 sqlfrom = "history"
439 if output:
440 sqlfrom = "history LEFT JOIN output_history USING (session, line)"
441 toget = "history.%s, output_history.output" % toget
442 if latest:
443 toget += ", MAX(session * 128 * 1024 + line)"
444 this_querry = "SELECT session, line, {} FROM {} ".format(toget, sqlfrom) + sql
445 cur = self.db.execute(this_querry, params)
446 if latest:
447 cur = (row[:-1] for row in cur)
448 if output: # Regroup into 3-tuples, and parse JSON
449 return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
450 return cur
452 @only_when_enabled
453 @catch_corrupt_db
454 def get_session_info(
455 self, session: int
456 ) -> tuple[int, datetime.datetime, datetime.datetime | None, int | None, str]:
457 """Get info about a session.
459 Parameters
460 ----------
461 session : int
462 Session number to retrieve.
464 Returns
465 -------
466 session_id : int
467 Session ID number
468 start : datetime
469 Timestamp for the start of the session.
470 end : datetime
471 Timestamp for the end of the session, or None if IPython crashed.
472 num_cmds : int
473 Number of commands run, or None if IPython crashed.
474 remark : str
475 A manually set description.
476 """
477 query = "SELECT * from sessions where session == ?"
478 return self.db.execute(query, (session,)).fetchone()
480 @catch_corrupt_db
481 def get_last_session_id(self) -> int | None:
482 """Get the last session ID currently in the database.
484 Within IPython, this should be the same as the value stored in
485 :attr:`HistoryManager.session_number`.
486 """
487 for record in self.get_tail(n=1, include_latest=True):
488 return record[0]
489 return None
491 @catch_corrupt_db
492 def get_tail(
493 self,
494 n: int = 10,
495 raw: bool = True,
496 output: bool = False,
497 include_latest: bool = False,
498 ) -> Iterable[tuple[int, int, InOrInOut]]:
499 """Get the last n lines from the history database.
501 Parameters
502 ----------
503 n : int
504 The number of lines to get
505 raw, output : bool
506 See :meth:`get_range`
507 include_latest : bool
508 If False (default), n+1 lines are fetched, and the latest one
509 is discarded. This is intended to be used where the function
510 is called by a user command, which it should not return.
512 Returns
513 -------
514 Tuples as :meth:`get_range`
515 """
516 self.writeout_cache()
517 if not include_latest:
518 n += 1
519 cur = self._run_sql(
520 "ORDER BY session DESC, line DESC LIMIT ?", (n,), raw=raw, output=output
521 )
522 if not include_latest:
523 return reversed(list(cur)[1:])
524 return reversed(list(cur))
526 @catch_corrupt_db
527 def search(
528 self,
529 pattern: str = "*",
530 raw: bool = True,
531 search_raw: bool = True,
532 output: bool = False,
533 n: int | None = None,
534 unique: bool = False,
535 ) -> Iterable[tuple[int, int, InOrInOut]]:
536 """Search the database using unix glob-style matching (wildcards
537 * and ?).
539 Parameters
540 ----------
541 pattern : str
542 The wildcarded pattern to match when searching
543 search_raw : bool
544 If True, search the raw input, otherwise, the parsed input
545 raw, output : bool
546 See :meth:`get_range`
547 n : None or int
548 If an integer is given, it defines the limit of
549 returned entries.
550 unique : bool
551 When it is true, return only unique entries.
553 Returns
554 -------
555 Tuples as :meth:`get_range`
556 """
557 tosearch = "source_raw" if search_raw else "source"
558 if output:
559 tosearch = "history." + tosearch
560 self.writeout_cache()
561 sqlform = "WHERE %s GLOB ?" % tosearch
562 params: tuple[typing.Any, ...] = (pattern,)
563 if unique:
564 sqlform += f" GROUP BY {tosearch}"
565 if n is not None:
566 sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
567 params += (n,)
568 elif unique:
569 sqlform += " ORDER BY session, line"
570 cur = self._run_sql(sqlform, params, raw=raw, output=output, latest=unique)
571 if n is not None:
572 return reversed(list(cur))
573 return cur
575 @catch_corrupt_db
576 def get_range(
577 self,
578 session: int,
579 start: int = 1,
580 stop: int | None = None,
581 raw: bool = True,
582 output: bool = False,
583 ) -> Iterable[tuple[int, int, InOrInOut]]:
584 """Retrieve input by session.
586 Parameters
587 ----------
588 session : int
589 Session number to retrieve.
590 start : int
591 First line to retrieve.
592 stop : int
593 End of line range (excluded from output itself). If None, retrieve
594 to the end of the session.
595 raw : bool
596 If True, return untranslated input
597 output : bool
598 If True, attempt to include output. This will be 'real' Python
599 objects for the current session, or text reprs from previous
600 sessions if db_log_output was enabled at the time. Where no output
601 is found, None is used.
603 Returns
604 -------
605 entries
606 An iterator over the desired lines. Each line is a 3-tuple, either
607 (session, line, input) if output is False, or
608 (session, line, (input, output)) if output is True.
609 """
610 params: tuple[typing.Any, ...]
611 if stop:
612 lineclause = "line >= ? AND line < ?"
613 params = (session, start, stop)
614 else:
615 lineclause = "line>=?"
616 params = (session, start)
618 return self._run_sql(
619 "WHERE session==? AND %s" % lineclause, params, raw=raw, output=output
620 )
622 def get_range_by_str(
623 self, rangestr: str, raw: bool = True, output: bool = False
624 ) -> Iterable[tuple[int, int, InOrInOut]]:
625 """Get lines of history from a string of ranges, as used by magic
626 commands %hist, %save, %macro, etc.
628 Parameters
629 ----------
630 rangestr : str
631 A string specifying ranges, e.g. "5 ~2/1-4". If empty string is used,
632 this will return everything from current session's history.
634 See the documentation of :func:`%history` for the full details.
636 raw, output : bool
637 As :meth:`get_range`
639 Returns
640 -------
641 Tuples as :meth:`get_range`
642 """
643 for sess, s, e in extract_hist_ranges(rangestr):
644 yield from self.get_range(sess, s, e, raw=raw, output=output)
647@dataclass
648class HistoryOutput:
649 output_type: typing.Literal[
650 "out_stream", "err_stream", "display_data", "execute_result"
651 ]
652 bundle: dict[str, str | list[str]]
655class HistoryManager(HistoryAccessor):
656 """A class to organize all history-related functionality in one place."""
658 # Public interface
660 # An instance of the IPython shell we are attached to
661 shell = Instance(
662 "IPython.core.interactiveshell.InteractiveShellABC", allow_none=False
663 )
664 # Lists to hold processed and raw history. These start with a blank entry
665 # so that we can index them starting from 1
666 input_hist_parsed = List([""])
667 input_hist_raw = List([""])
668 # A list of directories visited during session
669 dir_hist: List = List()
671 @default("dir_hist")
672 def _dir_hist_default(self) -> list[Path]:
673 try:
674 return [Path.cwd()]
675 except OSError:
676 return []
678 # A dict of output history, keyed with ints from the shell's
679 # execution count.
680 output_hist = Dict()
681 # The text/plain repr of outputs.
682 output_hist_reprs: dict[int, str] = Dict() # type: ignore [assignment]
683 # Maps execution_count to MIME bundles
684 outputs: dict[int, list[HistoryOutput]] = defaultdict(list)
685 # Maps execution_count to exception tracebacks
686 exceptions: dict[int, dict[str, Any]] = Dict() # type: ignore [assignment]
688 # The number of the current session in the history database
689 session_number: int = Integer() # type: ignore [assignment]
691 db_log_output = Bool(
692 False, help="Should the history database include output? (default: no)"
693 ).tag(config=True)
694 db_cache_size = Integer(
695 0,
696 help="Write to database every x commands (higher values save disk access & power).\n"
697 "Values of 1 or less effectively disable caching.",
698 ).tag(config=True)
699 # The input and output caches
700 db_input_cache: List[tuple[int, str, str]] = List()
701 db_output_cache: List[tuple[int, str]] = List()
703 # History saving in separate thread
704 save_thread = Instance("IPython.core.history.HistorySavingThread", allow_none=True)
706 @property
707 def save_flag(self) -> threading.Event | None:
708 if self.save_thread is not None:
709 return self.save_thread.save_flag
710 return None
712 # Private interface
713 # Variables used to store the three last inputs from the user. On each new
714 # history update, we populate the user's namespace with these, shifted as
715 # necessary.
716 _i00 = Unicode("")
717 _i = Unicode("")
718 _ii = Unicode("")
719 _iii = Unicode("")
721 # A regex matching all forms of the exit command, so that we don't store
722 # them in the history (it's annoying to rewind the first entry and land on
723 # an exit call).
724 _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
726 _instances: WeakSet[HistoryManager] = WeakSet()
727 _max_inst: int | float = float("inf")
729 def __init__(
730 self,
731 shell: InteractiveShell,
732 config: Configuration | None = None,
733 **traits: typing.Any,
734 ):
735 """Create a new history manager associated with a shell instance."""
736 super().__init__(shell=shell, config=config, **traits)
737 self.db_input_cache_lock = threading.Lock()
738 self.db_output_cache_lock = threading.Lock()
740 try:
741 self.new_session()
742 except OperationalError:
743 self.log.error(
744 "Failed to create history session in %s. History will not be saved.",
745 self.hist_file,
746 exc_info=True,
747 )
748 self._switch_to_memory_history()
750 self.using_thread = False
751 if self.enabled and self.hist_file != ":memory:":
752 self.save_thread = HistorySavingThread(self)
753 try:
754 self.save_thread.start()
755 except RuntimeError:
756 self.log.error(
757 "Failed to start history saving thread. History will not be saved.",
758 exc_info=True,
759 )
760 self._switch_to_memory_history()
761 self.save_thread = None
762 else:
763 self.using_thread = True
764 self._instances.add(self)
765 assert len(HistoryManager._instances) <= HistoryManager._max_inst, (
766 len(HistoryManager._instances),
767 HistoryManager._max_inst,
768 )
770 def _switch_to_memory_history(self) -> None:
771 """Switch history storage to an in-memory SQLite database."""
772 try:
773 self.db.close()
774 except Exception:
775 pass
776 self.hist_file = ":memory:"
777 self.init_db()
778 self.new_session()
780 def _stop_save_thread(self) -> None:
781 """Stop the background saving thread, if one is running.
783 The thread closes its own database connection as it exits, so this is
784 also what releases that connection.
785 """
786 if self.save_thread is not None:
787 self.save_thread.stop()
788 self.save_thread = None
790 def close(self) -> None:
791 """Stop the saving thread and close the database connection.
793 This is the deterministic counterpart to relying on garbage
794 collection: it shuts the saving thread down (which closes its private
795 connection) and then closes the manager's own connection. Safe to call
796 more than once.
797 """
798 self._stop_save_thread()
799 super().close()
801 def __del__(self) -> None:
802 self._stop_save_thread()
804 @classmethod
805 def _stop_thread(cls) -> None:
806 # Used before forking so the thread isn't running at fork
807 for inst in cls._instances:
808 inst._stop_save_thread()
810 def _restart_thread_if_stopped(self) -> None:
811 # Start the thread again after it was stopped for forking
812 if self.save_thread is None and self.using_thread:
813 self.save_thread = HistorySavingThread(self)
814 self.save_thread.start()
816 def _get_hist_file_name(self, profile: str | None = None) -> Path:
817 """Get default history file name based on the Shell's profile.
819 The profile parameter is ignored, but must exist for compatibility with
820 the parent class."""
821 profile_dir = self.shell.profile_dir.location
822 return Path(profile_dir) / "history.sqlite"
824 @only_when_enabled
825 def new_session(self, conn: sqlite3.Connection | None = None) -> None:
826 """Get a new session number."""
827 if conn is None:
828 conn = self.db
830 with conn:
831 cur = conn.execute(
832 """INSERT INTO sessions VALUES (NULL, ?, NULL,
833 NULL, '') """,
834 (datetime.datetime.now().isoformat(" "),),
835 )
836 assert isinstance(cur.lastrowid, int)
837 self.session_number = cur.lastrowid
839 def end_session(self) -> None:
840 """Close the database session, filling in the end time and line count."""
841 self.writeout_cache()
842 with self.db:
843 self.db.execute(
844 """UPDATE sessions SET end=?, num_cmds=? WHERE
845 session==?""",
846 (
847 datetime.datetime.now(datetime.UTC).isoformat(" "),
848 len(self.input_hist_parsed) - 1,
849 self.session_number,
850 ),
851 )
852 self.session_number = 0
854 def name_session(self, name: str) -> None:
855 """Give the current session a name in the history database."""
856 warn(
857 "name_session is deprecated in IPython 9.0 and will be removed in future versions",
858 DeprecationWarning,
859 stacklevel=2,
860 )
861 with self.db:
862 self.db.execute(
863 "UPDATE sessions SET remark=? WHERE session==?",
864 (name, self.session_number),
865 )
867 def reset(self, new_session: bool = True) -> None:
868 """Clear the session history, releasing all object references, and
869 optionally open a new session."""
870 self.output_hist.clear()
871 self.outputs.clear()
872 self.exceptions.clear()
874 # The directory history can't be completely empty
875 self.dir_hist[:] = [Path.cwd()]
877 if new_session:
878 if self.session_number:
879 self.end_session()
880 self.input_hist_parsed[:] = [""]
881 self.input_hist_raw[:] = [""]
882 self.new_session()
884 # ------------------------------
885 # Methods for retrieving history
886 # ------------------------------
887 def get_session_info(
888 self, session: int = 0
889 ) -> tuple[int, datetime.datetime, datetime.datetime | None, int | None, str]:
890 """Get info about a session.
892 Parameters
893 ----------
894 session : int
895 Session number to retrieve. The current session is 0, and negative
896 numbers count back from current session, so -1 is the previous session.
898 Returns
899 -------
900 session_id : int
901 Session ID number
902 start : datetime
903 Timestamp for the start of the session.
904 end : datetime
905 Timestamp for the end of the session, or None if IPython crashed.
906 num_cmds : int
907 Number of commands run, or None if IPython crashed.
908 remark : str
909 A manually set description.
910 """
911 if session <= 0:
912 session += self.session_number
914 return super().get_session_info(session=session)
916 @catch_corrupt_db
917 def get_tail(
918 self,
919 n: int = 10,
920 raw: bool = True,
921 output: bool = False,
922 include_latest: bool = False,
923 ) -> Iterable[tuple[int, int, InOrInOut]]:
924 """Get the last n lines from the history database.
926 Most recent entry last.
928 Completion will be reordered so that that the last ones are when
929 possible from current session.
931 Parameters
932 ----------
933 n : int
934 The number of lines to get
935 raw, output : bool
936 See :meth:`get_range`
937 include_latest : bool
938 If False (default), n+1 lines are fetched, and the latest one
939 is discarded. This is intended to be used where the function
940 is called by a user command, which it should not return.
942 Returns
943 -------
944 Tuples as :meth:`get_range`
945 """
946 self.writeout_cache()
947 if not include_latest:
948 n += 1
949 # cursor/line/entry
950 this_cur = list(
951 self._run_sql(
952 "WHERE session == ? ORDER BY line DESC LIMIT ? ",
953 (self.session_number, n),
954 raw=raw,
955 output=output,
956 )
957 )
958 other_cur = list(
959 self._run_sql(
960 "WHERE session != ? ORDER BY session DESC, line DESC LIMIT ?",
961 (self.session_number, n),
962 raw=raw,
963 output=output,
964 )
965 )
967 everything: list[tuple[int, int, InOrInOut]] = this_cur + other_cur
969 everything = everything[:n]
971 if not include_latest:
972 return list(everything)[:0:-1]
973 return list(everything)[::-1]
975 def _get_range_session(
976 self,
977 start: int = 1,
978 stop: int | None = None,
979 raw: bool = True,
980 output: bool = False,
981 ) -> Iterable[tuple[int, int, InOrInOut]]:
982 """Get input and output history from the current session. Called by
983 get_range, and takes similar parameters."""
984 input_hist = self.input_hist_raw if raw else self.input_hist_parsed
986 n = len(input_hist)
987 if start < 0:
988 start += n
989 if not stop or (stop > n):
990 stop = n
991 elif stop < 0:
992 stop += n
993 line: InOrInOut
994 for i in range(start, stop):
995 if output:
996 line = (input_hist[i], self.output_hist_reprs.get(i))
997 else:
998 line = input_hist[i]
999 yield (0, i, line)
1001 def get_range(
1002 self,
1003 session: int = 0,
1004 start: int = 1,
1005 stop: int | None = None,
1006 raw: bool = True,
1007 output: bool = False,
1008 ) -> Iterable[tuple[int, int, InOrInOut]]:
1009 """Retrieve input by session.
1011 Parameters
1012 ----------
1013 session : int
1014 Session number to retrieve. The current session is 0, and negative
1015 numbers count back from current session, so -1 is previous session.
1016 start : int
1017 First line to retrieve.
1018 stop : int
1019 End of line range (excluded from output itself). If None, retrieve
1020 to the end of the session.
1021 raw : bool
1022 If True, return untranslated input
1023 output : bool
1024 If True, attempt to include output. This will be 'real' Python
1025 objects for the current session, or text reprs from previous
1026 sessions if db_log_output was enabled at the time. Where no output
1027 is found, None is used.
1029 Returns
1030 -------
1031 entries
1032 An iterator over the desired lines. Each line is a 3-tuple, either
1033 (session, line, input) if output is False, or
1034 (session, line, (input, output)) if output is True.
1035 """
1036 if session <= 0:
1037 session += self.session_number
1038 if session == self.session_number: # Current session
1039 return self._get_range_session(start, stop, raw, output)
1040 return super().get_range(session, start, stop, raw, output)
1042 ## ----------------------------
1043 ## Methods for storing history:
1044 ## ----------------------------
1045 def store_inputs(
1046 self, line_num: int, source: str, source_raw: str | None = None
1047 ) -> None:
1048 """Store source and raw input in history and create input cache
1049 variables ``_i*``.
1051 Parameters
1052 ----------
1053 line_num : int
1054 The prompt number of this input.
1055 source : str
1056 Python input.
1057 source_raw : str, optional
1058 If given, this is the raw input without any IPython transformations
1059 applied to it. If not given, ``source`` is used.
1060 """
1061 if source_raw is None:
1062 source_raw = source
1063 source = source.rstrip("\n")
1064 source_raw = source_raw.rstrip("\n")
1066 # do not store exit/quit commands
1067 if self._exit_re.match(source_raw.strip()):
1068 return
1070 self.input_hist_parsed.append(source)
1071 self.input_hist_raw.append(source_raw)
1073 with self.db_input_cache_lock:
1074 self.db_input_cache.append((line_num, source, source_raw))
1075 # Trigger to flush cache and write to DB.
1076 if len(self.db_input_cache) >= self.db_cache_size:
1077 if self.using_thread:
1078 self._restart_thread_if_stopped()
1079 if self.save_flag is not None:
1080 self.save_flag.set()
1082 # update the auto _i variables
1083 self._iii = self._ii
1084 self._ii = self._i
1085 self._i = self._i00
1086 self._i00 = source_raw
1088 # hackish access to user namespace to create _i1,_i2... dynamically
1089 new_i = "_i%s" % line_num
1090 to_main = {"_i": self._i, "_ii": self._ii, "_iii": self._iii, new_i: self._i00}
1092 if self.shell is not None:
1093 self.shell.push(to_main, interactive=False)
1095 def store_output(self, line_num: int) -> None:
1096 """If database output logging is enabled, this saves all the
1097 outputs from the indicated prompt number to the database. It's
1098 called by run_cell after code has been executed.
1100 Parameters
1101 ----------
1102 line_num : int
1103 The line number from which to save outputs
1104 """
1105 if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
1106 return
1107 lnum: int = line_num
1108 output = self.output_hist_reprs[line_num]
1110 with self.db_output_cache_lock:
1111 self.db_output_cache.append((line_num, output))
1112 if self.db_cache_size <= 1 and self.using_thread:
1113 self._restart_thread_if_stopped()
1114 if self.save_flag is not None:
1115 self.save_flag.set()
1117 def _writeout_input_cache(self, conn: sqlite3.Connection) -> None:
1118 with conn:
1119 for line in self.db_input_cache:
1120 conn.execute(
1121 "INSERT INTO history VALUES (?, ?, ?, ?)",
1122 (self.session_number,) + line,
1123 )
1125 def _writeout_output_cache(self, conn: sqlite3.Connection) -> None:
1126 with conn:
1127 for line in self.db_output_cache:
1128 conn.execute(
1129 "INSERT INTO output_history VALUES (?, ?, ?)",
1130 (self.session_number,) + line,
1131 )
1133 @only_when_enabled
1134 def writeout_cache(self, conn: sqlite3.Connection | None = None) -> None:
1135 """Write any entries in the cache to the database."""
1136 if conn is None:
1137 conn = self.db
1139 with self.db_input_cache_lock:
1140 try:
1141 self._writeout_input_cache(conn)
1142 except sqlite3.IntegrityError:
1143 self.new_session(conn)
1144 print(
1145 "ERROR! Session/line number was not unique in",
1146 "database. History logging moved to new session",
1147 self.session_number,
1148 )
1149 try:
1150 # Try writing to the new session. If this fails, don't
1151 # recurse
1152 self._writeout_input_cache(conn)
1153 except sqlite3.IntegrityError:
1154 pass
1155 finally:
1156 self.db_input_cache = []
1158 with self.db_output_cache_lock:
1159 try:
1160 self._writeout_output_cache(conn)
1161 except sqlite3.IntegrityError:
1162 print(
1163 "!! Session/line number for output was not unique",
1164 "in database. Output will not be stored.",
1165 )
1166 finally:
1167 self.db_output_cache = []
1170if hasattr(os, "register_at_fork"):
1171 os.register_at_fork(before=HistoryManager._stop_thread)
1174from collections.abc import Callable, Iterator
1175from weakref import ReferenceType
1178@contextmanager
1179def hold(ref: ReferenceType[HistoryManager]) -> Iterator[ReferenceType[HistoryManager]]:
1180 """
1181 Context manger that hold a reference to a weak ref to make sure it
1182 is not GC'd during it's context.
1183 """
1184 r = ref()
1185 yield ref
1186 del r
1189class HistorySavingThread(threading.Thread):
1190 """This thread takes care of writing history to the database, so that
1191 the UI isn't held up while that happens.
1193 It waits for the HistoryManager's save_flag to be set, then writes out
1194 the history cache. The main thread is responsible for setting the flag when
1195 the cache size reaches a defined threshold."""
1197 save_flag: threading.Event
1198 daemon: bool = True
1199 _stop_now: bool = False
1200 enabled: bool = True
1201 history_manager: ref[HistoryManager]
1202 _stopped = False
1203 db: sqlite3.Connection | None = None
1205 def __init__(self, history_manager: HistoryManager) -> None:
1206 super().__init__(name="IPythonHistorySavingThread")
1207 self.history_manager = ref(history_manager)
1208 self.enabled = history_manager.enabled
1209 self.save_flag = threading.Event()
1211 @only_when_enabled
1212 def run(self) -> None:
1213 atexit.register(self.stop)
1214 # We need a separate db connection per thread:
1215 self.db = None
1216 try:
1217 hm: ReferenceType[HistoryManager]
1218 with hold(self.history_manager) as hm:
1219 if hm() is not None:
1220 self.db = sqlite3.connect(
1221 str(hm().hist_file), # type: ignore [union-attr]
1222 **cast(dict[str, t.Any], hm().connection_options), # type: ignore [union-attr]
1223 )
1224 while True:
1225 self.save_flag.wait()
1226 with hold(self.history_manager) as hm:
1227 if hm() is None:
1228 self._stop_now = True
1229 if self._stop_now:
1230 return
1231 self.save_flag.clear()
1232 if hm() is not None and self.db is not None:
1233 hm().writeout_cache(self.db) # type: ignore [union-attr]
1235 except Exception as e:
1236 print(
1237 (
1238 "The history saving thread hit an unexpected error (%s)."
1239 "History will not be written to the database."
1240 )
1241 % repr(e)
1242 )
1243 finally:
1244 # Always close our per-thread connection, whatever path we exit by
1245 # (normal stop, a dropped HistoryManager, or an unexpected error).
1246 # Leaving it open lets the sqlite3.Connection be garbage collected
1247 # unclosed, which raises a spurious ``ResourceWarning`` in whatever
1248 # code happens to be running when the collection occurs.
1249 if self.db is not None:
1250 self.db.close()
1251 self.db = None
1252 atexit.unregister(self.stop)
1254 def stop(self) -> None:
1255 """This can be called from the main thread to safely stop this thread.
1257 Note that it does not attempt to write out remaining history before
1258 exiting. That should be done by calling the HistoryManager's
1259 end_session method."""
1260 if self._stopped:
1261 return
1262 self._stop_now = True
1264 self.save_flag.set()
1265 self._stopped = True
1266 if self.ident is not None and self != threading.current_thread():
1267 self.join()
1269 def __del__(self) -> None:
1270 self.stop()
1273# To match, e.g. ~5/8-~2/3, or ~4 (without trailing slash for full session)
1274# Session numbers: ~N or N/
1275# Line numbers: N (just digits, no ~)
1276# Range syntax: 4-6 (with end) or 4- (without end, means "onward")
1277range_re = re.compile(
1278 r"""
1279((?P<startsess>(?:~?\d+/)))?
1280(?P<start>\d+)?
1281((?P<sep>[\-:])
1282 ((?P<endsess>(?:~?\d+/)))?
1283 (?P<end>\d*))?
1284$""",
1285 re.VERBOSE,
1286)
1289def extract_hist_ranges(ranges_str: str) -> Iterable[tuple[int, int, int | None]]:
1290 """Turn a string of history ranges into 3-tuples of (session, start, stop).
1292 Empty string results in a `[(0, 1, None)]`, i.e. "everything from current
1293 session".
1295 Examples
1296 --------
1297 >>> list(extract_hist_ranges("~8/5-~7/4 2"))
1298 [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
1299 >>> list(extract_hist_ranges("~4/"))
1300 [(-4, 1, None)]
1301 >>> list(extract_hist_ranges("4-"))
1302 [(0, 4, None)]
1303 >>> list(extract_hist_ranges("~4/4-"))
1304 [(-4, 4, None)]
1305 """
1306 if ranges_str == "":
1307 yield (0, 1, None) # Everything from current session
1308 return
1310 for range_str in ranges_str.split():
1311 rmatch = range_re.match(range_str)
1312 if not rmatch:
1313 continue
1314 start = rmatch.group("start")
1315 sep = rmatch.group("sep")
1316 if start:
1317 start = int(start)
1318 end = rmatch.group("end")
1319 if sep == "-":
1320 end = (int(end) + 1) if end else None
1321 else:
1322 end = int(end) if end else start + 1
1323 else:
1324 if not rmatch.group("startsess"):
1325 continue
1326 start = 1
1327 end = None
1328 startsess = rmatch.group("startsess") or "0"
1329 endsess = rmatch.group("endsess") or startsess
1330 startsess = startsess.rstrip("/")
1331 endsess = endsess.rstrip("/")
1332 startsess = int(startsess.replace("~", "-"))
1333 endsess = int(endsess.replace("~", "-"))
1334 assert endsess >= startsess, "start session must be earlier than end session"
1336 if endsess == startsess:
1337 yield (startsess, start, end)
1338 continue
1339 # Multiple sessions in one range:
1340 yield (startsess, start, None)
1341 for sess in range(startsess + 1, endsess):
1342 yield (sess, 1, None)
1343 yield (endsess, 1, end)
1346def _format_lineno(session: int, line: int) -> str:
1347 """Helper function to format line numbers properly."""
1348 if session == 0:
1349 return str(line)
1350 return "{}#{}".format(session, line)