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
48from collections.abc import Callable, Iterator
49from weakref import ReferenceType
52if TYPE_CHECKING:
53 import sqlite3
54 from types import TracebackType
56 from IPython.core.interactiveshell import InteractiveShell
57 from traitlets.config import Config as Configuration
59# sqlite3 is optional: it is a pure-Python package wrapping the `_sqlite3`
60# extension module, and CPython can be built (or packaged) without the latter.
61# Importing it costs ~8 ms and 23 modules, and a session that never touches
62# history never needs it, so everything below resolves it on first use.
63#
64# Note that `importlib.util.find_spec("sqlite3")` is *not* a valid
65# availability check: the pure-Python package is on disk either way, and only
66# the import of `_sqlite3` underneath it fails. Only trying the import tells
67# the truth -- and it must catch `ImportError`, not just `ModuleNotFoundError`,
68# since the extension can also be present but fail to load.
71@functools.cache
72def _sqlite3() -> t.Any:
73 """Return the `sqlite3` module, with IPython's converter registered.
75 Raises `ImportError` if this Python has no working sqlite3.
76 """
77 import sqlite3
79 sqlite3.register_converter(
80 "timestamp", lambda val: datetime.datetime.fromisoformat(val.decode())
81 )
82 return sqlite3
85@functools.cache
86def _sqlite3_found() -> bool:
87 """Whether this Python can actually import sqlite3."""
88 try:
89 _sqlite3()
90 except ImportError:
91 return False
92 return True
95@functools.cache
96def _db_errors() -> tuple[type[BaseException], ...]:
97 """The sqlite3 errors to catch, or an empty tuple if it is unavailable.
99 An empty tuple in an `except` clause simply never matches, which is the
100 right behaviour when there is no database to fail in the first place.
101 """
102 if not _sqlite3_found():
103 return ()
104 sqlite3 = _sqlite3()
105 return (sqlite3.DatabaseError, sqlite3.OperationalError)
108@functools.cache
109def _operational_error() -> tuple[type[BaseException], ...]:
110 """`sqlite3.OperationalError`, or an empty tuple if unavailable."""
111 if not _sqlite3_found():
112 return ()
113 return (_sqlite3().OperationalError,)
116InOrInOut = str | tuple[str, str | None]
118# -----------------------------------------------------------------------------
119# Classes and functions
120# -----------------------------------------------------------------------------
123@undoc
124class DummyDB:
125 """Dummy DB that will act as a black hole for history.
127 Only used in the absence of sqlite"""
129 def execute(*args: typing.Any, **kwargs: typing.Any) -> list:
130 return []
132 def commit(self, *args: typing.Any, **kwargs: typing.Any) -> None:
133 pass
135 def __enter__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
136 pass
138 def __exit__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
139 pass
141 def close(self, *args: typing.Any, **kwargs: typing.Any) -> None:
142 pass
145_P = ParamSpec("_P")
146_R = t.TypeVar("_R")
149def only_when_enabled(f: t.Callable[_P, _R]) -> t.Callable[_P, _R]:
150 """Decorator: return an empty list in the absence of sqlite.
152 Typed as signature-preserving (like the ``decorator``-package version it
153 replaces): the empty-list fallback for a disabled accessor is invisible to
154 the type system, as before.
155 """
157 @functools.wraps(f)
158 def wrapper(*a: _P.args, **kw: _P.kwargs) -> _R:
159 self = cast("HistoryAccessor", a[0])
160 if not self.enabled:
161 return cast(_R, [])
162 else:
163 return f(*a, **kw)
165 return wrapper
168# use 16kB as threshold for whether a corrupt history db should be saved
169# that should be at least 100 entries or so
170_SAVE_DB_SIZE = 16384
173def catch_corrupt_db(f: t.Callable[_P, _R]) -> t.Callable[_P, _R]:
174 """A decorator which wraps HistoryAccessor method calls to catch errors from
175 a corrupt SQLite database, move the old database out of the way, and create
176 a new one.
178 We avoid clobbering larger databases because this may be triggered due to filesystem issues,
179 not just a corrupt file.
180 """
182 @functools.wraps(f)
183 def wrapper(*a: _P.args, **kw: _P.kwargs) -> _R:
184 self = cast("HistoryAccessor", a[0])
185 try:
186 return f(*a, **kw)
187 except _db_errors() as e:
188 self._corrupt_db_counter += 1
189 self.log.error("Failed to open SQLite history %s (%s).", self.hist_file, e)
190 if self.hist_file != ":memory:":
191 if self._corrupt_db_counter > self._corrupt_db_limit:
192 self.hist_file = ":memory:"
193 self.log.error(
194 "Failed to load history too many times, history will not be saved."
195 )
196 elif self.hist_file.is_file():
197 # move the file out of the way
198 base = str(self.hist_file.parent / self.hist_file.stem)
199 ext = self.hist_file.suffix
200 size = self.hist_file.stat().st_size
201 if size >= _SAVE_DB_SIZE:
202 # if there's significant content, avoid clobbering
203 now = (
204 datetime.datetime.now(datetime.UTC)
205 .isoformat()
206 .replace(":", ".")
207 )
208 newpath = base + "-corrupt-" + now + ext
209 # don't clobber previous corrupt backups
210 for i in range(100):
211 if not Path(newpath).exists():
212 break
213 else:
214 newpath = base + "-corrupt-" + now + ("-%i" % i) + ext
215 else:
216 # not much content, possibly empty; don't worry about clobbering
217 # maybe we should just delete it?
218 newpath = base + "-corrupt" + ext
219 self.hist_file.rename(newpath)
220 self.log.error(
221 "History file was moved to %s and a new file created.", newpath
222 )
223 self.init_db()
224 return cast(_R, [])
225 else:
226 # Failed with :memory:, something serious is wrong
227 raise
229 return wrapper
232class HistoryAccessorBase(LoggingConfigurable):
233 """An abstract class for History Accessors"""
235 def get_tail(
236 self,
237 n: int = 10,
238 raw: bool = True,
239 output: bool = False,
240 include_latest: bool = False,
241 ) -> Iterable[tuple[int, int, InOrInOut]]:
242 raise NotImplementedError
244 def search(
245 self,
246 pattern: str = "*",
247 raw: bool = True,
248 search_raw: bool = True,
249 output: bool = False,
250 n: int | None = None,
251 unique: bool = False,
252 ) -> Iterable[tuple[int, int, InOrInOut]]:
253 raise NotImplementedError
255 def get_range(
256 self,
257 session: int,
258 start: int = 1,
259 stop: int | None = None,
260 raw: bool = True,
261 output: bool = False,
262 ) -> Iterable[tuple[int, int, InOrInOut]]:
263 raise NotImplementedError
265 def get_range_by_str(
266 self, rangestr: str, raw: bool = True, output: bool = False
267 ) -> Iterable[tuple[int, int, InOrInOut]]:
268 raise NotImplementedError
271class HistoryAccessor(HistoryAccessorBase):
272 """Access the history database without adding to it.
274 This is intended for use by standalone history tools. IPython shells use
275 HistoryManager, below, which is a subclass of this."""
277 # counter for init_db retries, so we don't keep trying over and over
278 _corrupt_db_counter = 0
279 # after two failures, fallback on :memory:
280 _corrupt_db_limit = 2
282 # String holding the path to the history file
283 hist_file = Union(
284 [Instance(Path), Unicode()],
285 help="""Path to file to use for SQLite history database.
287 By default, IPython will put the history database in the IPython
288 profile directory. If you would rather share one history among
289 profiles, you can set this value in each, so that they are consistent.
291 Due to an issue with fcntl, SQLite is known to misbehave on some NFS
292 mounts. If you see IPython hanging, try setting this to something on a
293 local disk, e.g::
295 ipython --HistoryManager.hist_file=/tmp/ipython_hist.sqlite
297 you can also use the specific value `:memory:` (including the colon
298 at both end but not the back ticks), to avoid creating an history file.
300 """,
301 ).tag(config=True)
303 enabled = Bool(
304 help="""enable the SQLite history
306 set enabled=False to disable the SQLite history,
307 in which case there will be no stored history, no SQLite connection,
308 and no background saving thread. This may be necessary in some
309 threaded environments where IPython is embedded.
310 """,
311 ).tag(config=True)
313 @default("enabled")
314 def _enabled_default(self) -> bool:
315 # dynamic rather than `Bool(_sqlite3_found())`: a static default is
316 # evaluated when the class is created, which would import sqlite3 on
317 # every `import IPython`
318 return _sqlite3_found()
320 connection_options = Dict(
321 help="""Options for configuring the SQLite connection
323 These options are passed as keyword args to sqlite3.connect
324 when establishing database connections.
325 """
326 ).tag(config=True)
328 @default("connection_options")
329 def _default_connection_options(self) -> dict[str, bool]:
330 return dict(check_same_thread=False)
332 # The SQLite database
333 db = Any()
335 @observe("db")
336 @only_when_enabled
337 def _db_changed(self, change): # type: ignore [no-untyped-def]
338 """validate the db, since it can be an Instance of two different types"""
339 new = change["new"]
340 connection_types = (DummyDB, _sqlite3().Connection)
341 if not isinstance(new, connection_types):
342 msg = "{}.db must be sqlite3 Connection or DummyDB, not {!r}".format(
343 self.__class__.__name__,
344 new,
345 )
346 raise TraitError(msg)
348 def __init__(
349 self, profile: str = "default", hist_file: str = "", **traits: typing.Any
350 ) -> None:
351 """Create a new history accessor.
353 Parameters
354 ----------
355 profile : str
356 The name of the profile from which to open history.
357 hist_file : str
358 Path to an SQLite history database stored by IPython. If specified,
359 hist_file overrides profile.
360 config : :class:`~traitlets.config.loader.Config`
361 Config object. hist_file can also be set through this.
362 """
363 super().__init__(**traits)
364 # defer setting hist_file from kwarg until after init,
365 # otherwise the default kwarg value would clobber any value
366 # set by config
367 if hist_file:
368 self.hist_file = hist_file
370 try:
371 self.hist_file
372 except TraitError:
373 # No one has set the hist_file, yet.
374 self.hist_file = self._get_hist_file_name(profile)
376 self.init_db()
378 def _get_hist_file_name(self, profile: str = "default") -> Path:
379 """Find the history file for the given profile name.
381 This is overridden by the HistoryManager subclass, to use the shell's
382 active profile.
384 Parameters
385 ----------
386 profile : str
387 The name of a profile which has a history file.
388 """
389 return Path(locate_profile(profile)) / "history.sqlite"
391 @catch_corrupt_db
392 def init_db(self) -> None:
393 """Connect to the database, and create tables if necessary."""
394 if not self.enabled:
395 self.db = DummyDB()
396 self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db)
397 return
399 # use detect_types so that timestamps return datetime objects
400 sqlite3 = _sqlite3()
401 kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES)
402 kwargs.update(self.connection_options)
403 self.db = sqlite3.connect(str(self.hist_file), **kwargs)
404 self._finalizer = weakref.finalize(self, lambda db: db.close(), self.db)
405 with self.db:
406 self.db.execute(
407 """CREATE TABLE IF NOT EXISTS sessions (session integer
408 primary key autoincrement, start timestamp,
409 end timestamp, num_cmds integer, remark text)"""
410 )
411 self.db.execute(
412 """CREATE TABLE IF NOT EXISTS history
413 (session integer, line integer, source text, source_raw text,
414 PRIMARY KEY (session, line))"""
415 )
416 # Output history is optional, but ensure the table's there so it can be
417 # enabled later.
418 self.db.execute(
419 """CREATE TABLE IF NOT EXISTS output_history
420 (session integer, line integer, output text,
421 PRIMARY KEY (session, line))"""
422 )
423 # success! reset corrupt db count
424 self._corrupt_db_counter = 0
426 def close(self) -> None:
427 """Close the SQLite database connection.
429 Prefer calling this to closing ``self.db`` directly: it gives
430 subclasses (notably :class:`HistoryManager`) a single place to hook in
431 the rest of their teardown. Safe to call more than once.
432 """
433 self.db.close()
435 def __enter__(self) -> HistoryAccessor:
436 """Support use as a context manager for deterministic cleanup::
438 with HistoryAccessor(hist_file=path) as history:
439 ...
440 # connection is closed here
442 :class:`HistoryManager` additionally stops its saving thread on exit.
443 """
444 return self
446 def __exit__(
447 self,
448 exc_type: type[BaseException] | None,
449 exc_value: BaseException | None,
450 traceback: TracebackType | None,
451 ) -> None:
452 self.close()
454 def writeout_cache(self) -> None:
455 """Overridden by HistoryManager to dump the cache before certain
456 database lookups."""
457 pass
459 ## -------------------------------
460 ## Methods for retrieving history:
461 ## -------------------------------
462 def _run_sql(
463 self,
464 sql: str,
465 params: tuple,
466 raw: bool = True,
467 output: bool = False,
468 latest: bool = False,
469 ) -> Iterable[tuple[int, int, InOrInOut]]:
470 """Prepares and runs an SQL query for the history database.
472 Parameters
473 ----------
474 sql : str
475 Any filtering expressions to go after SELECT ... FROM ...
476 params : tuple
477 Parameters passed to the SQL query (to replace "?")
478 raw, output : bool
479 See :meth:`get_range`
480 latest : bool
481 Select rows with max (session, line)
483 Returns
484 -------
485 Tuples as :meth:`get_range`
486 """
487 toget = "source_raw" if raw else "source"
488 sqlfrom = "history"
489 if output:
490 sqlfrom = "history LEFT JOIN output_history USING (session, line)"
491 toget = "history.%s, output_history.output" % toget
492 if latest:
493 toget += ", MAX(session * 128 * 1024 + line)"
494 this_querry = "SELECT session, line, {} FROM {} ".format(toget, sqlfrom) + sql
495 cur = self.db.execute(this_querry, params)
496 if latest:
497 cur = (row[:-1] for row in cur)
498 if output: # Regroup into 3-tuples, and parse JSON
499 return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
500 return cur
502 @only_when_enabled
503 @catch_corrupt_db
504 def get_session_info(
505 self, session: int
506 ) -> tuple[int, datetime.datetime, datetime.datetime | None, int | None, str]:
507 """Get info about a session.
509 Parameters
510 ----------
511 session : int
512 Session number to retrieve.
514 Returns
515 -------
516 session_id : int
517 Session ID number
518 start : datetime
519 Timestamp for the start of the session.
520 end : datetime
521 Timestamp for the end of the session, or None if IPython crashed.
522 num_cmds : int
523 Number of commands run, or None if IPython crashed.
524 remark : str
525 A manually set description.
526 """
527 query = "SELECT * from sessions where session == ?"
528 return self.db.execute(query, (session,)).fetchone()
530 @catch_corrupt_db
531 def get_last_session_id(self) -> int | None:
532 """Get the last session ID currently in the database.
534 Within IPython, this should be the same as the value stored in
535 :attr:`HistoryManager.session_number`.
536 """
537 for record in self.get_tail(n=1, include_latest=True):
538 return record[0]
539 return None
541 @catch_corrupt_db
542 def get_tail(
543 self,
544 n: int = 10,
545 raw: bool = True,
546 output: bool = False,
547 include_latest: bool = False,
548 ) -> Iterable[tuple[int, int, InOrInOut]]:
549 """Get the last n lines from the history database.
551 Parameters
552 ----------
553 n : int
554 The number of lines to get
555 raw, output : bool
556 See :meth:`get_range`
557 include_latest : bool
558 If False (default), n+1 lines are fetched, and the latest one
559 is discarded. This is intended to be used where the function
560 is called by a user command, which it should not return.
562 Returns
563 -------
564 Tuples as :meth:`get_range`
565 """
566 self.writeout_cache()
567 if not include_latest:
568 n += 1
569 cur = self._run_sql(
570 "ORDER BY session DESC, line DESC LIMIT ?", (n,), raw=raw, output=output
571 )
572 if not include_latest:
573 return reversed(list(cur)[1:])
574 return reversed(list(cur))
576 @catch_corrupt_db
577 def search(
578 self,
579 pattern: str = "*",
580 raw: bool = True,
581 search_raw: bool = True,
582 output: bool = False,
583 n: int | None = None,
584 unique: bool = False,
585 ) -> Iterable[tuple[int, int, InOrInOut]]:
586 """Search the database using unix glob-style matching (wildcards
587 * and ?).
589 Parameters
590 ----------
591 pattern : str
592 The wildcarded pattern to match when searching
593 search_raw : bool
594 If True, search the raw input, otherwise, the parsed input
595 raw, output : bool
596 See :meth:`get_range`
597 n : None or int
598 If an integer is given, it defines the limit of
599 returned entries.
600 unique : bool
601 When it is true, return only unique entries.
603 Returns
604 -------
605 Tuples as :meth:`get_range`
606 """
607 tosearch = "source_raw" if search_raw else "source"
608 if output:
609 tosearch = "history." + tosearch
610 self.writeout_cache()
611 sqlform = "WHERE %s GLOB ?" % tosearch
612 params: tuple[typing.Any, ...] = (pattern,)
613 if unique:
614 sqlform += f" GROUP BY {tosearch}"
615 if n is not None:
616 sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
617 params += (n,)
618 elif unique:
619 sqlform += " ORDER BY session, line"
620 cur = self._run_sql(sqlform, params, raw=raw, output=output, latest=unique)
621 if n is not None:
622 return reversed(list(cur))
623 return cur
625 @catch_corrupt_db
626 def get_range(
627 self,
628 session: int,
629 start: int = 1,
630 stop: int | None = None,
631 raw: bool = True,
632 output: bool = False,
633 ) -> Iterable[tuple[int, int, InOrInOut]]:
634 """Retrieve input by session.
636 Parameters
637 ----------
638 session : int
639 Session number to retrieve.
640 start : int
641 First line to retrieve.
642 stop : int
643 End of line range (excluded from output itself). If None, retrieve
644 to the end of the session.
645 raw : bool
646 If True, return untranslated input
647 output : bool
648 If True, attempt to include output. This will be 'real' Python
649 objects for the current session, or text reprs from previous
650 sessions if db_log_output was enabled at the time. Where no output
651 is found, None is used.
653 Returns
654 -------
655 entries
656 An iterator over the desired lines. Each line is a 3-tuple, either
657 (session, line, input) if output is False, or
658 (session, line, (input, output)) if output is True.
659 """
660 params: tuple[typing.Any, ...]
661 if stop:
662 lineclause = "line >= ? AND line < ?"
663 params = (session, start, stop)
664 else:
665 lineclause = "line>=?"
666 params = (session, start)
668 return self._run_sql(
669 "WHERE session==? AND %s" % lineclause, params, raw=raw, output=output
670 )
672 def get_range_by_str(
673 self, rangestr: str, raw: bool = True, output: bool = False
674 ) -> Iterable[tuple[int, int, InOrInOut]]:
675 """Get lines of history from a string of ranges, as used by magic
676 commands %hist, %save, %macro, etc.
678 Parameters
679 ----------
680 rangestr : str
681 A string specifying ranges, e.g. "5 ~2/1-4". If empty string is used,
682 this will return everything from current session's history.
684 See the documentation of :func:`%history` for the full details.
686 raw, output : bool
687 As :meth:`get_range`
689 Returns
690 -------
691 Tuples as :meth:`get_range`
692 """
693 for sess, s, e in extract_hist_ranges(rangestr):
694 yield from self.get_range(sess, s, e, raw=raw, output=output)
697@dataclass
698class HistoryOutput:
699 output_type: typing.Literal[
700 "out_stream", "err_stream", "display_data", "execute_result"
701 ]
702 bundle: dict[str, str | list[str]]
705class HistoryManager(HistoryAccessor):
706 """A class to organize all history-related functionality in one place."""
708 # Public interface
710 # An instance of the IPython shell we are attached to
711 shell = Instance(
712 "IPython.core.interactiveshell.InteractiveShellABC", allow_none=False
713 )
714 # Lists to hold processed and raw history. These start with a blank entry
715 # so that we can index them starting from 1
716 input_hist_parsed = List([""])
717 input_hist_raw = List([""])
718 # A list of directories visited during session
719 dir_hist: List = List()
721 @default("dir_hist")
722 def _dir_hist_default(self) -> list[Path]:
723 try:
724 return [Path.cwd()]
725 except OSError:
726 return []
728 # A dict of output history, keyed with ints from the shell's
729 # execution count.
730 output_hist = Dict()
731 # The text/plain repr of outputs.
732 output_hist_reprs: dict[int, str] = Dict() # type: ignore [assignment]
733 # Maps execution_count to MIME bundles
734 outputs: dict[int, list[HistoryOutput]] = defaultdict(list)
735 # Maps execution_count to exception tracebacks
736 exceptions: dict[int, dict[str, Any]] = Dict() # type: ignore [assignment]
738 # The number of the current session in the history database
739 session_number: int = Integer() # type: ignore [assignment]
741 db_log_output = Bool(
742 False, help="Should the history database include output? (default: no)"
743 ).tag(config=True)
744 db_cache_size = Integer(
745 0,
746 help="Write to database every x commands (higher values save disk access & power).\n"
747 "Values of 1 or less effectively disable caching.",
748 ).tag(config=True)
749 # The input and output caches
750 db_input_cache: List[tuple[int, str, str]] = List()
751 db_output_cache: List[tuple[int, str]] = List()
753 # History saving in separate thread
754 save_thread = Instance("IPython.core.history.HistorySavingThread", allow_none=True)
756 @property
757 def save_flag(self) -> threading.Event | None:
758 if self.save_thread is not None:
759 return self.save_thread.save_flag
760 return None
762 # Private interface
763 # Variables used to store the three last inputs from the user. On each new
764 # history update, we populate the user's namespace with these, shifted as
765 # necessary.
766 _i00 = Unicode("")
767 _i = Unicode("")
768 _ii = Unicode("")
769 _iii = Unicode("")
771 # A regex matching all forms of the exit command, so that we don't store
772 # them in the history (it's annoying to rewind the first entry and land on
773 # an exit call).
774 _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
776 _instances: WeakSet[HistoryManager] = WeakSet()
777 _max_inst: int | float = float("inf")
779 def __init__(
780 self,
781 shell: InteractiveShell,
782 config: Configuration | None = None,
783 **traits: typing.Any,
784 ):
785 """Create a new history manager associated with a shell instance."""
786 super().__init__(shell=shell, config=config, **traits)
787 self.db_input_cache_lock = threading.Lock()
788 self.db_output_cache_lock = threading.Lock()
790 try:
791 self.new_session()
792 except _operational_error():
793 self.log.error(
794 "Failed to create history session in %s. History will not be saved.",
795 self.hist_file,
796 exc_info=True,
797 )
798 self._switch_to_memory_history()
800 self.using_thread = False
801 if self.enabled and self.hist_file != ":memory:":
802 self.save_thread = HistorySavingThread(self)
803 try:
804 self.save_thread.start()
805 except RuntimeError:
806 self.log.error(
807 "Failed to start history saving thread. History will not be saved.",
808 exc_info=True,
809 )
810 self._switch_to_memory_history()
811 self.save_thread = None
812 else:
813 self.using_thread = True
814 self._instances.add(self)
815 assert len(HistoryManager._instances) <= HistoryManager._max_inst, (
816 len(HistoryManager._instances),
817 HistoryManager._max_inst,
818 )
820 def _switch_to_memory_history(self) -> None:
821 """Switch history storage to an in-memory SQLite database."""
822 try:
823 self.db.close()
824 except Exception:
825 pass
826 self.hist_file = ":memory:"
827 self.init_db()
828 self.new_session()
830 def _stop_save_thread(self) -> None:
831 """Stop the background saving thread, if one is running.
833 The thread closes its own database connection as it exits, so this is
834 also what releases that connection.
835 """
836 if self.save_thread is not None:
837 self.save_thread.stop()
838 self.save_thread = None
840 def close(self) -> None:
841 """Stop the saving thread and close the database connection.
843 This is the deterministic counterpart to relying on garbage
844 collection: it shuts the saving thread down (which closes its private
845 connection) and then closes the manager's own connection. Safe to call
846 more than once.
847 """
848 self._stop_save_thread()
849 super().close()
851 def __del__(self) -> None:
852 self._stop_save_thread()
854 @classmethod
855 def _stop_thread(cls) -> None:
856 # Used before forking so the thread isn't running at fork
857 for inst in cls._instances:
858 inst._stop_save_thread()
860 def _restart_thread_if_stopped(self) -> None:
861 # Start the thread again after it was stopped for forking
862 if self.save_thread is None and self.using_thread:
863 self.save_thread = HistorySavingThread(self)
864 self.save_thread.start()
866 def _get_hist_file_name(self, profile: str | None = None) -> Path:
867 """Get default history file name based on the Shell's profile.
869 The profile parameter is ignored, but must exist for compatibility with
870 the parent class."""
871 profile_dir = self.shell.profile_dir.location
872 return Path(profile_dir) / "history.sqlite"
874 @only_when_enabled
875 def new_session(self, conn: sqlite3.Connection | None = None) -> None:
876 """Get a new session number."""
877 if conn is None:
878 conn = self.db
880 with conn:
881 cur = conn.execute(
882 """INSERT INTO sessions VALUES (NULL, ?, NULL,
883 NULL, '') """,
884 (datetime.datetime.now().isoformat(" "),),
885 )
886 assert isinstance(cur.lastrowid, int)
887 self.session_number = cur.lastrowid
889 def end_session(self) -> None:
890 """Close the database session, filling in the end time and line count."""
891 self.writeout_cache()
892 with self.db:
893 self.db.execute(
894 """UPDATE sessions SET end=?, num_cmds=? WHERE
895 session==?""",
896 (
897 datetime.datetime.now(datetime.UTC).isoformat(" "),
898 len(self.input_hist_parsed) - 1,
899 self.session_number,
900 ),
901 )
902 self.session_number = 0
904 def name_session(self, name: str) -> None:
905 """Give the current session a name in the history database."""
906 warn(
907 "name_session is deprecated in IPython 9.0 and will be removed in future versions",
908 DeprecationWarning,
909 stacklevel=2,
910 )
911 with self.db:
912 self.db.execute(
913 "UPDATE sessions SET remark=? WHERE session==?",
914 (name, self.session_number),
915 )
917 def reset(self, new_session: bool = True) -> None:
918 """Clear the session history, releasing all object references, and
919 optionally open a new session."""
920 self.output_hist.clear()
921 self.outputs.clear()
922 self.exceptions.clear()
924 # The directory history can't be completely empty
925 self.dir_hist[:] = [Path.cwd()]
927 if new_session:
928 if self.session_number:
929 self.end_session()
930 self.input_hist_parsed[:] = [""]
931 self.input_hist_raw[:] = [""]
932 self.new_session()
934 # ------------------------------
935 # Methods for retrieving history
936 # ------------------------------
937 def get_session_info(
938 self, session: int = 0
939 ) -> tuple[int, datetime.datetime, datetime.datetime | None, int | None, str]:
940 """Get info about a session.
942 Parameters
943 ----------
944 session : int
945 Session number to retrieve. The current session is 0, and negative
946 numbers count back from current session, so -1 is the previous session.
948 Returns
949 -------
950 session_id : int
951 Session ID number
952 start : datetime
953 Timestamp for the start of the session.
954 end : datetime
955 Timestamp for the end of the session, or None if IPython crashed.
956 num_cmds : int
957 Number of commands run, or None if IPython crashed.
958 remark : str
959 A manually set description.
960 """
961 if session <= 0:
962 session += self.session_number
964 return super().get_session_info(session=session)
966 @catch_corrupt_db
967 def get_tail(
968 self,
969 n: int = 10,
970 raw: bool = True,
971 output: bool = False,
972 include_latest: bool = False,
973 ) -> Iterable[tuple[int, int, InOrInOut]]:
974 """Get the last n lines from the history database.
976 Most recent entry last.
978 Completion will be reordered so that that the last ones are when
979 possible from current session.
981 Parameters
982 ----------
983 n : int
984 The number of lines to get
985 raw, output : bool
986 See :meth:`get_range`
987 include_latest : bool
988 If False (default), n+1 lines are fetched, and the latest one
989 is discarded. This is intended to be used where the function
990 is called by a user command, which it should not return.
992 Returns
993 -------
994 Tuples as :meth:`get_range`
995 """
996 self.writeout_cache()
997 if not include_latest:
998 n += 1
999 # cursor/line/entry
1000 this_cur = list(
1001 self._run_sql(
1002 "WHERE session == ? ORDER BY line DESC LIMIT ? ",
1003 (self.session_number, n),
1004 raw=raw,
1005 output=output,
1006 )
1007 )
1008 other_cur = list(
1009 self._run_sql(
1010 "WHERE session != ? ORDER BY session DESC, line DESC LIMIT ?",
1011 (self.session_number, n),
1012 raw=raw,
1013 output=output,
1014 )
1015 )
1017 everything: list[tuple[int, int, InOrInOut]] = this_cur + other_cur
1019 everything = everything[:n]
1021 if not include_latest:
1022 return list(everything)[:0:-1]
1023 return list(everything)[::-1]
1025 def _get_range_session(
1026 self,
1027 start: int = 1,
1028 stop: int | None = None,
1029 raw: bool = True,
1030 output: bool = False,
1031 ) -> Iterable[tuple[int, int, InOrInOut]]:
1032 """Get input and output history from the current session. Called by
1033 get_range, and takes similar parameters."""
1034 input_hist = self.input_hist_raw if raw else self.input_hist_parsed
1036 n = len(input_hist)
1037 if start < 0:
1038 start += n
1039 if not stop or (stop > n):
1040 stop = n
1041 elif stop < 0:
1042 stop += n
1043 line: InOrInOut
1044 for i in range(start, stop):
1045 if output:
1046 line = (input_hist[i], self.output_hist_reprs.get(i))
1047 else:
1048 line = input_hist[i]
1049 yield (0, i, line)
1051 def get_range(
1052 self,
1053 session: int = 0,
1054 start: int = 1,
1055 stop: int | None = None,
1056 raw: bool = True,
1057 output: bool = False,
1058 ) -> Iterable[tuple[int, int, InOrInOut]]:
1059 """Retrieve input by session.
1061 Parameters
1062 ----------
1063 session : int
1064 Session number to retrieve. The current session is 0, and negative
1065 numbers count back from current session, so -1 is previous session.
1066 start : int
1067 First line to retrieve.
1068 stop : int
1069 End of line range (excluded from output itself). If None, retrieve
1070 to the end of the session.
1071 raw : bool
1072 If True, return untranslated input
1073 output : bool
1074 If True, attempt to include output. This will be 'real' Python
1075 objects for the current session, or text reprs from previous
1076 sessions if db_log_output was enabled at the time. Where no output
1077 is found, None is used.
1079 Returns
1080 -------
1081 entries
1082 An iterator over the desired lines. Each line is a 3-tuple, either
1083 (session, line, input) if output is False, or
1084 (session, line, (input, output)) if output is True.
1085 """
1086 if session <= 0:
1087 session += self.session_number
1088 if session == self.session_number: # Current session
1089 return self._get_range_session(start, stop, raw, output)
1090 return super().get_range(session, start, stop, raw, output)
1092 ## ----------------------------
1093 ## Methods for storing history:
1094 ## ----------------------------
1095 def store_inputs(
1096 self, line_num: int, source: str, source_raw: str | None = None
1097 ) -> None:
1098 """Store source and raw input in history and create input cache
1099 variables ``_i*``.
1101 Parameters
1102 ----------
1103 line_num : int
1104 The prompt number of this input.
1105 source : str
1106 Python input.
1107 source_raw : str, optional
1108 If given, this is the raw input without any IPython transformations
1109 applied to it. If not given, ``source`` is used.
1110 """
1111 if source_raw is None:
1112 source_raw = source
1113 source = source.rstrip("\n")
1114 source_raw = source_raw.rstrip("\n")
1116 # do not store exit/quit commands
1117 if self._exit_re.match(source_raw.strip()):
1118 return
1120 self.input_hist_parsed.append(source)
1121 self.input_hist_raw.append(source_raw)
1123 with self.db_input_cache_lock:
1124 self.db_input_cache.append((line_num, source, source_raw))
1125 # Trigger to flush cache and write to DB.
1126 if len(self.db_input_cache) >= self.db_cache_size:
1127 if self.using_thread:
1128 self._restart_thread_if_stopped()
1129 if self.save_flag is not None:
1130 self.save_flag.set()
1132 # update the auto _i variables
1133 self._iii = self._ii
1134 self._ii = self._i
1135 self._i = self._i00
1136 self._i00 = source_raw
1138 # hackish access to user namespace to create _i1,_i2... dynamically
1139 new_i = "_i%s" % line_num
1140 to_main = {"_i": self._i, "_ii": self._ii, "_iii": self._iii, new_i: self._i00}
1142 if self.shell is not None:
1143 self.shell.push(to_main, interactive=False)
1145 def store_output(self, line_num: int) -> None:
1146 """If database output logging is enabled, this saves all the
1147 outputs from the indicated prompt number to the database. It's
1148 called by run_cell after code has been executed.
1150 Parameters
1151 ----------
1152 line_num : int
1153 The line number from which to save outputs
1154 """
1155 if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
1156 return
1157 lnum: int = line_num
1158 output = self.output_hist_reprs[line_num]
1160 with self.db_output_cache_lock:
1161 self.db_output_cache.append((line_num, output))
1162 if self.db_cache_size <= 1 and self.using_thread:
1163 self._restart_thread_if_stopped()
1164 if self.save_flag is not None:
1165 self.save_flag.set()
1167 def _writeout_input_cache(self, conn: sqlite3.Connection) -> None:
1168 with conn:
1169 for line in self.db_input_cache:
1170 conn.execute(
1171 "INSERT INTO history VALUES (?, ?, ?, ?)",
1172 (self.session_number,) + line,
1173 )
1175 def _writeout_output_cache(self, conn: sqlite3.Connection) -> None:
1176 with conn:
1177 for line in self.db_output_cache:
1178 conn.execute(
1179 "INSERT INTO output_history VALUES (?, ?, ?)",
1180 (self.session_number,) + line,
1181 )
1183 @only_when_enabled
1184 def writeout_cache(self, conn: sqlite3.Connection | None = None) -> None:
1185 """Write any entries in the cache to the database."""
1186 if conn is None:
1187 conn = self.db
1189 with self.db_input_cache_lock:
1190 try:
1191 self._writeout_input_cache(conn)
1192 except _sqlite3().IntegrityError:
1193 self.new_session(conn)
1194 print(
1195 "ERROR! Session/line number was not unique in",
1196 "database. History logging moved to new session",
1197 self.session_number,
1198 )
1199 try:
1200 # Try writing to the new session. If this fails, don't
1201 # recurse
1202 self._writeout_input_cache(conn)
1203 except _sqlite3().IntegrityError:
1204 pass
1205 finally:
1206 self.db_input_cache = []
1208 with self.db_output_cache_lock:
1209 try:
1210 self._writeout_output_cache(conn)
1211 except _sqlite3().IntegrityError:
1212 print(
1213 "!! Session/line number for output was not unique",
1214 "in database. Output will not be stored.",
1215 )
1216 finally:
1217 self.db_output_cache = []
1220if hasattr(os, "register_at_fork"):
1221 os.register_at_fork(before=HistoryManager._stop_thread)
1225@contextmanager
1226def hold(ref: ReferenceType[HistoryManager]) -> Iterator[ReferenceType[HistoryManager]]:
1227 """
1228 Context manger that hold a reference to a weak ref to make sure it
1229 is not GC'd during it's context.
1230 """
1231 r = ref()
1232 yield ref
1233 del r
1236class HistorySavingThread(threading.Thread):
1237 """This thread takes care of writing history to the database, so that
1238 the UI isn't held up while that happens.
1240 It waits for the HistoryManager's save_flag to be set, then writes out
1241 the history cache. The main thread is responsible for setting the flag when
1242 the cache size reaches a defined threshold."""
1244 save_flag: threading.Event
1245 daemon: bool = True
1246 _stop_now: bool = False
1247 enabled: bool = True
1248 history_manager: ref[HistoryManager]
1249 _stopped = False
1250 db: sqlite3.Connection | None = None
1252 def __init__(self, history_manager: HistoryManager) -> None:
1253 super().__init__(name="IPythonHistorySavingThread")
1254 self.history_manager = ref(history_manager)
1255 self.enabled = history_manager.enabled
1256 self.save_flag = threading.Event()
1258 @only_when_enabled
1259 def run(self) -> None:
1260 atexit.register(self.stop)
1261 # We need a separate db connection per thread:
1262 self.db = None
1263 try:
1264 hm: ReferenceType[HistoryManager]
1265 with hold(self.history_manager) as hm:
1266 if hm() is not None:
1267 self.db = _sqlite3().connect(
1268 str(hm().hist_file), # type: ignore [union-attr]
1269 **cast(dict[str, t.Any], hm().connection_options), # type: ignore [union-attr]
1270 )
1271 while True:
1272 self.save_flag.wait()
1273 with hold(self.history_manager) as hm:
1274 if hm() is None:
1275 self._stop_now = True
1276 if self._stop_now:
1277 return
1278 self.save_flag.clear()
1279 if hm() is not None and self.db is not None:
1280 hm().writeout_cache(self.db) # type: ignore [union-attr]
1282 except Exception as e:
1283 print(
1284 (
1285 "The history saving thread hit an unexpected error (%s)."
1286 "History will not be written to the database."
1287 )
1288 % repr(e)
1289 )
1290 finally:
1291 # Always close our per-thread connection, whatever path we exit by
1292 # (normal stop, a dropped HistoryManager, or an unexpected error).
1293 # Leaving it open lets the sqlite3.Connection be garbage collected
1294 # unclosed, which raises a spurious ``ResourceWarning`` in whatever
1295 # code happens to be running when the collection occurs.
1296 if self.db is not None:
1297 self.db.close()
1298 self.db = None
1299 atexit.unregister(self.stop)
1301 def stop(self) -> None:
1302 """This can be called from the main thread to safely stop this thread.
1304 Note that it does not attempt to write out remaining history before
1305 exiting. That should be done by calling the HistoryManager's
1306 end_session method."""
1307 if self._stopped:
1308 return
1309 self._stop_now = True
1311 self.save_flag.set()
1312 self._stopped = True
1313 if self.ident is not None and self != threading.current_thread():
1314 self.join()
1316 def __del__(self) -> None:
1317 self.stop()
1320# To match, e.g. ~5/8-~2/3, or ~4 (without trailing slash for full session)
1321# Session numbers: ~N or N/
1322# Line numbers: N (just digits, no ~)
1323# Range syntax: 4-6 (with end) or 4- (without end, means "onward")
1324range_re = re.compile(
1325 r"""
1326((?P<startsess>(?:~?\d+/)))?
1327(?P<start>\d+)?
1328((?P<sep>[\-:])
1329 ((?P<endsess>(?:~?\d+/)))?
1330 (?P<end>\d*))?
1331$""",
1332 re.VERBOSE,
1333)
1336def extract_hist_ranges(ranges_str: str) -> Iterable[tuple[int, int, int | None]]:
1337 """Turn a string of history ranges into 3-tuples of (session, start, stop).
1339 Empty string results in a `[(0, 1, None)]`, i.e. "everything from current
1340 session".
1342 Examples
1343 --------
1344 >>> list(extract_hist_ranges("~8/5-~7/4 2"))
1345 [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
1346 >>> list(extract_hist_ranges("~4/"))
1347 [(-4, 1, None)]
1348 >>> list(extract_hist_ranges("4-"))
1349 [(0, 4, None)]
1350 >>> list(extract_hist_ranges("~4/4-"))
1351 [(-4, 4, None)]
1352 """
1353 if ranges_str == "":
1354 yield (0, 1, None) # Everything from current session
1355 return
1357 for range_str in ranges_str.split():
1358 rmatch = range_re.match(range_str)
1359 if not rmatch:
1360 continue
1361 start = rmatch.group("start")
1362 sep = rmatch.group("sep")
1363 if start:
1364 start = int(start)
1365 end = rmatch.group("end")
1366 if sep == "-":
1367 end = (int(end) + 1) if end else None
1368 else:
1369 end = int(end) if end else start + 1
1370 else:
1371 if not rmatch.group("startsess"):
1372 continue
1373 start = 1
1374 end = None
1375 startsess = rmatch.group("startsess") or "0"
1376 endsess = rmatch.group("endsess") or startsess
1377 startsess = startsess.rstrip("/")
1378 endsess = endsess.rstrip("/")
1379 startsess = int(startsess.replace("~", "-"))
1380 endsess = int(endsess.replace("~", "-"))
1381 assert endsess >= startsess, "start session must be earlier than end session"
1383 if endsess == startsess:
1384 yield (startsess, start, end)
1385 continue
1386 # Multiple sessions in one range:
1387 yield (startsess, start, None)
1388 for sess in range(startsess + 1, endsess):
1389 yield (sess, 1, None)
1390 yield (endsess, 1, end)
1393def _format_lineno(session: int, line: int) -> str:
1394 """Helper function to format line numbers properly."""
1395 if session == 0:
1396 return str(line)
1397 return "{}#{}".format(session, line)