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 | None = 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 | None = 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 or None
554 The number of lines to get. None gets the default of 10, which a
555 kernel ``history_request`` without ``n`` asks for.
556 raw, output : bool
557 See :meth:`get_range`
558 include_latest : bool
559 If False (default), n+1 lines are fetched, and the latest one
560 is discarded. This is intended to be used where the function
561 is called by a user command, which it should not return.
563 Returns
564 -------
565 Tuples as :meth:`get_range`
566 """
567 if n is None:
568 n = 10
569 self.writeout_cache()
570 if not include_latest:
571 n += 1
572 cur = self._run_sql(
573 "ORDER BY session DESC, line DESC LIMIT ?", (n,), raw=raw, output=output
574 )
575 if not include_latest:
576 return reversed(list(cur)[1:])
577 return reversed(list(cur))
579 @catch_corrupt_db
580 def search(
581 self,
582 pattern: str = "*",
583 raw: bool = True,
584 search_raw: bool = True,
585 output: bool = False,
586 n: int | None = None,
587 unique: bool = False,
588 ) -> Iterable[tuple[int, int, InOrInOut]]:
589 """Search the database using unix glob-style matching (wildcards
590 * and ?).
592 Parameters
593 ----------
594 pattern : str
595 The wildcarded pattern to match when searching
596 search_raw : bool
597 If True, search the raw input, otherwise, the parsed input
598 raw, output : bool
599 See :meth:`get_range`
600 n : None or int
601 If an integer is given, it defines the limit of
602 returned entries.
603 unique : bool
604 When it is true, return only unique entries.
606 Returns
607 -------
608 Tuples as :meth:`get_range`
609 """
610 tosearch = "source_raw" if search_raw else "source"
611 if output:
612 tosearch = "history." + tosearch
613 self.writeout_cache()
614 sqlform = "WHERE %s GLOB ?" % tosearch
615 params: tuple[typing.Any, ...] = (pattern,)
616 if unique:
617 sqlform += f" GROUP BY {tosearch}"
618 if n is not None:
619 sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
620 params += (n,)
621 elif unique:
622 sqlform += " ORDER BY session, line"
623 cur = self._run_sql(sqlform, params, raw=raw, output=output, latest=unique)
624 if n is not None:
625 return reversed(list(cur))
626 return cur
628 @catch_corrupt_db
629 def get_range(
630 self,
631 session: int,
632 start: int = 1,
633 stop: int | None = None,
634 raw: bool = True,
635 output: bool = False,
636 ) -> Iterable[tuple[int, int, InOrInOut]]:
637 """Retrieve input by session.
639 Parameters
640 ----------
641 session : int
642 Session number to retrieve.
643 start : int
644 First line to retrieve.
645 stop : int
646 End of line range (excluded from output itself). If None, retrieve
647 to the end of the session.
648 raw : bool
649 If True, return untranslated input
650 output : bool
651 If True, attempt to include output. This will be 'real' Python
652 objects for the current session, or text reprs from previous
653 sessions if db_log_output was enabled at the time. Where no output
654 is found, None is used.
656 Returns
657 -------
658 entries
659 An iterator over the desired lines. Each line is a 3-tuple, either
660 (session, line, input) if output is False, or
661 (session, line, (input, output)) if output is True.
662 """
663 params: tuple[typing.Any, ...]
664 if stop:
665 lineclause = "line >= ? AND line < ?"
666 params = (session, start, stop)
667 else:
668 lineclause = "line>=?"
669 params = (session, start)
671 return self._run_sql(
672 "WHERE session==? AND %s" % lineclause, params, raw=raw, output=output
673 )
675 def get_range_by_str(
676 self, rangestr: str, raw: bool = True, output: bool = False
677 ) -> Iterable[tuple[int, int, InOrInOut]]:
678 """Get lines of history from a string of ranges, as used by magic
679 commands %hist, %save, %macro, etc.
681 Parameters
682 ----------
683 rangestr : str
684 A string specifying ranges, e.g. "5 ~2/1-4". If empty string is used,
685 this will return everything from current session's history.
687 See the documentation of :func:`%history` for the full details.
689 raw, output : bool
690 As :meth:`get_range`
692 Returns
693 -------
694 Tuples as :meth:`get_range`
695 """
696 for sess, s, e in extract_hist_ranges(rangestr):
697 yield from self.get_range(sess, s, e, raw=raw, output=output)
700@dataclass
701class HistoryOutput:
702 output_type: typing.Literal[
703 "out_stream", "err_stream", "display_data", "execute_result"
704 ]
705 bundle: dict[str, str | list[str]]
708class HistoryManager(HistoryAccessor):
709 """A class to organize all history-related functionality in one place."""
711 # Public interface
713 # An instance of the IPython shell we are attached to
714 shell = Instance(
715 "IPython.core.interactiveshell.InteractiveShellABC", allow_none=False
716 )
717 # Lists to hold processed and raw history. These start with a blank entry
718 # so that we can index them starting from 1
719 input_hist_parsed = List([""])
720 input_hist_raw = List([""])
721 # A list of directories visited during session
722 dir_hist: List = List()
724 @default("dir_hist")
725 def _dir_hist_default(self) -> list[Path]:
726 try:
727 return [Path.cwd()]
728 except OSError:
729 return []
731 # A dict of output history, keyed with ints from the shell's
732 # execution count.
733 output_hist = Dict()
734 # The text/plain repr of outputs.
735 output_hist_reprs: dict[int, str] = Dict() # type: ignore [assignment]
736 # Maps execution_count to MIME bundles
737 outputs: dict[int, list[HistoryOutput]] = defaultdict(list)
738 # Maps execution_count to exception tracebacks
739 exceptions: dict[int, dict[str, Any]] = Dict() # type: ignore [assignment]
741 # The number of the current session in the history database
742 session_number: int = Integer() # type: ignore [assignment]
744 db_log_output = Bool(
745 False, help="Should the history database include output? (default: no)"
746 ).tag(config=True)
747 db_cache_size = Integer(
748 0,
749 help="Write to database every x commands (higher values save disk access & power).\n"
750 "Values of 1 or less effectively disable caching.",
751 ).tag(config=True)
752 # The input and output caches
753 db_input_cache: List[tuple[int, str, str]] = List()
754 db_output_cache: List[tuple[int, str]] = List()
756 # History saving in separate thread
757 save_thread = Instance("IPython.core.history.HistorySavingThread", allow_none=True)
759 @property
760 def save_flag(self) -> threading.Event | None:
761 if self.save_thread is not None:
762 return self.save_thread.save_flag
763 return None
765 # Private interface
766 # Variables used to store the three last inputs from the user. On each new
767 # history update, we populate the user's namespace with these, shifted as
768 # necessary.
769 _i00 = Unicode("")
770 _i = Unicode("")
771 _ii = Unicode("")
772 _iii = Unicode("")
774 # A regex matching all forms of the exit command, so that we don't store
775 # them in the history (it's annoying to rewind the first entry and land on
776 # an exit call).
777 _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
779 _instances: WeakSet[HistoryManager] = WeakSet()
780 _max_inst: int | float = float("inf")
782 def __init__(
783 self,
784 shell: InteractiveShell,
785 config: Configuration | None = None,
786 **traits: typing.Any,
787 ):
788 """Create a new history manager associated with a shell instance."""
789 super().__init__(shell=shell, config=config, **traits)
790 self.db_input_cache_lock = threading.Lock()
791 self.db_output_cache_lock = threading.Lock()
793 try:
794 self.new_session()
795 except _operational_error():
796 self.log.error(
797 "Failed to create history session in %s. History will not be saved.",
798 self.hist_file,
799 exc_info=True,
800 )
801 self._switch_to_memory_history()
803 self.using_thread = False
804 if self.enabled and self.hist_file != ":memory:":
805 self.save_thread = HistorySavingThread(self)
806 try:
807 self.save_thread.start()
808 except RuntimeError:
809 self.log.error(
810 "Failed to start history saving thread. History will not be saved.",
811 exc_info=True,
812 )
813 self._switch_to_memory_history()
814 self.save_thread = None
815 else:
816 self.using_thread = True
817 self._instances.add(self)
818 assert len(HistoryManager._instances) <= HistoryManager._max_inst, (
819 len(HistoryManager._instances),
820 HistoryManager._max_inst,
821 )
823 def _switch_to_memory_history(self) -> None:
824 """Switch history storage to an in-memory SQLite database."""
825 try:
826 self.db.close()
827 except Exception:
828 pass
829 self.hist_file = ":memory:"
830 self.init_db()
831 self.new_session()
833 def _stop_save_thread(self) -> None:
834 """Stop the background saving thread, if one is running.
836 The thread closes its own database connection as it exits, so this is
837 also what releases that connection.
838 """
839 if self.save_thread is not None:
840 self.save_thread.stop()
841 self.save_thread = None
843 def close(self) -> None:
844 """Stop the saving thread and close the database connection.
846 This is the deterministic counterpart to relying on garbage
847 collection: it shuts the saving thread down (which closes its private
848 connection) and then closes the manager's own connection. Safe to call
849 more than once.
850 """
851 self._stop_save_thread()
852 super().close()
854 def __del__(self) -> None:
855 self._stop_save_thread()
857 @classmethod
858 def _stop_thread(cls) -> None:
859 # Used before forking so the thread isn't running at fork
860 for inst in cls._instances:
861 inst._stop_save_thread()
863 def _restart_thread_if_stopped(self) -> None:
864 # Start the thread again after it was stopped for forking
865 if self.save_thread is None and self.using_thread:
866 self.save_thread = HistorySavingThread(self)
867 self.save_thread.start()
869 def _get_hist_file_name(self, profile: str | None = None) -> Path:
870 """Get default history file name based on the Shell's profile.
872 The profile parameter is ignored, but must exist for compatibility with
873 the parent class."""
874 profile_dir = self.shell.profile_dir.location
875 return Path(profile_dir) / "history.sqlite"
877 @only_when_enabled
878 def new_session(self, conn: sqlite3.Connection | None = None) -> None:
879 """Get a new session number."""
880 if conn is None:
881 conn = self.db
883 with conn:
884 cur = conn.execute(
885 """INSERT INTO sessions VALUES (NULL, ?, NULL,
886 NULL, '') """,
887 (datetime.datetime.now().isoformat(" "),),
888 )
889 assert isinstance(cur.lastrowid, int)
890 self.session_number = cur.lastrowid
892 def end_session(self) -> None:
893 """Close the database session, filling in the end time and line count."""
894 self.writeout_cache()
895 with self.db:
896 self.db.execute(
897 """UPDATE sessions SET end=?, num_cmds=? WHERE
898 session==?""",
899 (
900 datetime.datetime.now(datetime.UTC).isoformat(" "),
901 len(self.input_hist_parsed) - 1,
902 self.session_number,
903 ),
904 )
905 self.session_number = 0
907 def name_session(self, name: str) -> None:
908 """Give the current session a name in the history database."""
909 warn(
910 "name_session is deprecated in IPython 9.0 and will be removed in future versions",
911 DeprecationWarning,
912 stacklevel=2,
913 )
914 with self.db:
915 self.db.execute(
916 "UPDATE sessions SET remark=? WHERE session==?",
917 (name, self.session_number),
918 )
920 def reset(self, new_session: bool = True) -> None:
921 """Clear the session history, releasing all object references, and
922 optionally open a new session."""
923 self.output_hist.clear()
924 self.outputs.clear()
925 self.exceptions.clear()
927 # The directory history can't be completely empty
928 self.dir_hist[:] = [Path.cwd()]
930 if new_session:
931 if self.session_number:
932 self.end_session()
933 self.input_hist_parsed[:] = [""]
934 self.input_hist_raw[:] = [""]
935 self.new_session()
937 # ------------------------------
938 # Methods for retrieving history
939 # ------------------------------
940 def get_session_info(
941 self, session: int = 0
942 ) -> tuple[int, datetime.datetime, datetime.datetime | None, int | None, str]:
943 """Get info about a session.
945 Parameters
946 ----------
947 session : int
948 Session number to retrieve. The current session is 0, and negative
949 numbers count back from current session, so -1 is the previous session.
951 Returns
952 -------
953 session_id : int
954 Session ID number
955 start : datetime
956 Timestamp for the start of the session.
957 end : datetime
958 Timestamp for the end of the session, or None if IPython crashed.
959 num_cmds : int
960 Number of commands run, or None if IPython crashed.
961 remark : str
962 A manually set description.
963 """
964 if session <= 0:
965 session += self.session_number
967 return super().get_session_info(session=session)
969 @catch_corrupt_db
970 def get_tail(
971 self,
972 n: int | None = 10,
973 raw: bool = True,
974 output: bool = False,
975 include_latest: bool = False,
976 ) -> Iterable[tuple[int, int, InOrInOut]]:
977 """Get the last n lines from the history database.
979 Most recent entry last.
981 Completion will be reordered so that that the last ones are when
982 possible from current session.
984 Parameters
985 ----------
986 n : int or None
987 The number of lines to get. None gets the default of 10, which a
988 kernel ``history_request`` without ``n`` asks for.
989 raw, output : bool
990 See :meth:`get_range`
991 include_latest : bool
992 If False (default), n+1 lines are fetched, and the latest one
993 is discarded. This is intended to be used where the function
994 is called by a user command, which it should not return.
996 Returns
997 -------
998 Tuples as :meth:`get_range`
999 """
1000 if n is None:
1001 n = 10
1002 self.writeout_cache()
1003 if not include_latest:
1004 n += 1
1005 # cursor/line/entry
1006 this_cur = list(
1007 self._run_sql(
1008 "WHERE session == ? ORDER BY line DESC LIMIT ? ",
1009 (self.session_number, n),
1010 raw=raw,
1011 output=output,
1012 )
1013 )
1014 other_cur = list(
1015 self._run_sql(
1016 "WHERE session != ? ORDER BY session DESC, line DESC LIMIT ?",
1017 (self.session_number, n),
1018 raw=raw,
1019 output=output,
1020 )
1021 )
1023 everything: list[tuple[int, int, InOrInOut]] = this_cur + other_cur
1025 everything = everything[:n]
1027 if not include_latest:
1028 return list(everything)[:0:-1]
1029 return list(everything)[::-1]
1031 def _get_range_session(
1032 self,
1033 start: int = 1,
1034 stop: int | None = None,
1035 raw: bool = True,
1036 output: bool = False,
1037 ) -> Iterable[tuple[int, int, InOrInOut]]:
1038 """Get input and output history from the current session. Called by
1039 get_range, and takes similar parameters."""
1040 input_hist = self.input_hist_raw if raw else self.input_hist_parsed
1042 n = len(input_hist)
1043 if start < 0:
1044 start += n
1045 if not stop or (stop > n):
1046 stop = n
1047 elif stop < 0:
1048 stop += n
1049 line: InOrInOut
1050 for i in range(start, stop):
1051 if output:
1052 line = (input_hist[i], self.output_hist_reprs.get(i))
1053 else:
1054 line = input_hist[i]
1055 yield (0, i, line)
1057 def get_range(
1058 self,
1059 session: int = 0,
1060 start: int = 1,
1061 stop: int | None = None,
1062 raw: bool = True,
1063 output: bool = False,
1064 ) -> Iterable[tuple[int, int, InOrInOut]]:
1065 """Retrieve input by session.
1067 Parameters
1068 ----------
1069 session : int
1070 Session number to retrieve. The current session is 0, and negative
1071 numbers count back from current session, so -1 is previous session.
1072 start : int
1073 First line to retrieve.
1074 stop : int
1075 End of line range (excluded from output itself). If None, retrieve
1076 to the end of the session.
1077 raw : bool
1078 If True, return untranslated input
1079 output : bool
1080 If True, attempt to include output. This will be 'real' Python
1081 objects for the current session, or text reprs from previous
1082 sessions if db_log_output was enabled at the time. Where no output
1083 is found, None is used.
1085 Returns
1086 -------
1087 entries
1088 An iterator over the desired lines. Each line is a 3-tuple, either
1089 (session, line, input) if output is False, or
1090 (session, line, (input, output)) if output is True.
1091 """
1092 if session <= 0:
1093 session += self.session_number
1094 if session == self.session_number: # Current session
1095 return self._get_range_session(start, stop, raw, output)
1096 return super().get_range(session, start, stop, raw, output)
1098 ## ----------------------------
1099 ## Methods for storing history:
1100 ## ----------------------------
1101 def store_inputs(
1102 self, line_num: int, source: str, source_raw: str | None = None
1103 ) -> None:
1104 """Store source and raw input in history and create input cache
1105 variables ``_i*``.
1107 Parameters
1108 ----------
1109 line_num : int
1110 The prompt number of this input.
1111 source : str
1112 Python input.
1113 source_raw : str, optional
1114 If given, this is the raw input without any IPython transformations
1115 applied to it. If not given, ``source`` is used.
1116 """
1117 if source_raw is None:
1118 source_raw = source
1119 source = source.rstrip("\n")
1120 source_raw = source_raw.rstrip("\n")
1122 # do not store exit/quit commands
1123 if self._exit_re.match(source_raw.strip()):
1124 return
1126 self.input_hist_parsed.append(source)
1127 self.input_hist_raw.append(source_raw)
1129 with self.db_input_cache_lock:
1130 self.db_input_cache.append((line_num, source, source_raw))
1131 # Trigger to flush cache and write to DB.
1132 if len(self.db_input_cache) >= self.db_cache_size:
1133 if self.using_thread:
1134 self._restart_thread_if_stopped()
1135 if self.save_flag is not None:
1136 self.save_flag.set()
1138 # update the auto _i variables
1139 self._iii = self._ii
1140 self._ii = self._i
1141 self._i = self._i00
1142 self._i00 = source_raw
1144 # hackish access to user namespace to create _i1,_i2... dynamically
1145 new_i = "_i%s" % line_num
1146 to_main = {"_i": self._i, "_ii": self._ii, "_iii": self._iii, new_i: self._i00}
1148 if self.shell is not None:
1149 self.shell.push(to_main, interactive=False)
1151 def store_output(self, line_num: int) -> None:
1152 """If database output logging is enabled, this saves all the
1153 outputs from the indicated prompt number to the database. It's
1154 called by run_cell after code has been executed.
1156 Parameters
1157 ----------
1158 line_num : int
1159 The line number from which to save outputs
1160 """
1161 if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
1162 return
1163 lnum: int = line_num
1164 output = self.output_hist_reprs[line_num]
1166 with self.db_output_cache_lock:
1167 self.db_output_cache.append((line_num, output))
1168 if self.db_cache_size <= 1 and self.using_thread:
1169 self._restart_thread_if_stopped()
1170 if self.save_flag is not None:
1171 self.save_flag.set()
1173 def _writeout_input_cache(self, conn: sqlite3.Connection) -> None:
1174 with conn:
1175 for line in self.db_input_cache:
1176 conn.execute(
1177 "INSERT INTO history VALUES (?, ?, ?, ?)",
1178 (self.session_number,) + line,
1179 )
1181 def _writeout_output_cache(self, conn: sqlite3.Connection) -> None:
1182 with conn:
1183 for line in self.db_output_cache:
1184 conn.execute(
1185 "INSERT INTO output_history VALUES (?, ?, ?)",
1186 (self.session_number,) + line,
1187 )
1189 @only_when_enabled
1190 def writeout_cache(self, conn: sqlite3.Connection | None = None) -> None:
1191 """Write any entries in the cache to the database."""
1192 if conn is None:
1193 conn = self.db
1195 with self.db_input_cache_lock:
1196 try:
1197 self._writeout_input_cache(conn)
1198 except _sqlite3().IntegrityError:
1199 self.new_session(conn)
1200 print(
1201 "ERROR! Session/line number was not unique in",
1202 "database. History logging moved to new session",
1203 self.session_number,
1204 )
1205 try:
1206 # Try writing to the new session. If this fails, don't
1207 # recurse
1208 self._writeout_input_cache(conn)
1209 except _sqlite3().IntegrityError:
1210 pass
1211 finally:
1212 self.db_input_cache = []
1214 with self.db_output_cache_lock:
1215 try:
1216 self._writeout_output_cache(conn)
1217 except _sqlite3().IntegrityError:
1218 print(
1219 "!! Session/line number for output was not unique",
1220 "in database. Output will not be stored.",
1221 )
1222 finally:
1223 self.db_output_cache = []
1226if hasattr(os, "register_at_fork"):
1227 os.register_at_fork(before=HistoryManager._stop_thread)
1231@contextmanager
1232def hold(ref: ReferenceType[HistoryManager]) -> Iterator[ReferenceType[HistoryManager]]:
1233 """
1234 Context manger that hold a reference to a weak ref to make sure it
1235 is not GC'd during it's context.
1236 """
1237 r = ref()
1238 yield ref
1239 del r
1242class HistorySavingThread(threading.Thread):
1243 """This thread takes care of writing history to the database, so that
1244 the UI isn't held up while that happens.
1246 It waits for the HistoryManager's save_flag to be set, then writes out
1247 the history cache. The main thread is responsible for setting the flag when
1248 the cache size reaches a defined threshold."""
1250 save_flag: threading.Event
1251 daemon: bool = True
1252 _stop_now: bool = False
1253 enabled: bool = True
1254 history_manager: ref[HistoryManager]
1255 _stopped = False
1256 db: sqlite3.Connection | None = None
1258 def __init__(self, history_manager: HistoryManager) -> None:
1259 super().__init__(name="IPythonHistorySavingThread")
1260 self.history_manager = ref(history_manager)
1261 self.enabled = history_manager.enabled
1262 self.save_flag = threading.Event()
1264 @only_when_enabled
1265 def run(self) -> None:
1266 atexit.register(self.stop)
1267 # We need a separate db connection per thread:
1268 self.db = None
1269 try:
1270 hm: ReferenceType[HistoryManager]
1271 with hold(self.history_manager) as hm:
1272 if hm() is not None:
1273 self.db = _sqlite3().connect(
1274 str(hm().hist_file), # type: ignore [union-attr]
1275 **cast(dict[str, t.Any], hm().connection_options), # type: ignore [union-attr]
1276 )
1277 while True:
1278 self.save_flag.wait()
1279 with hold(self.history_manager) as hm:
1280 if hm() is None:
1281 self._stop_now = True
1282 if self._stop_now:
1283 return
1284 self.save_flag.clear()
1285 if hm() is not None and self.db is not None:
1286 hm().writeout_cache(self.db) # type: ignore [union-attr]
1288 except Exception as e:
1289 print(
1290 (
1291 "The history saving thread hit an unexpected error (%s)."
1292 "History will not be written to the database."
1293 )
1294 % repr(e)
1295 )
1296 finally:
1297 # Always close our per-thread connection, whatever path we exit by
1298 # (normal stop, a dropped HistoryManager, or an unexpected error).
1299 # Leaving it open lets the sqlite3.Connection be garbage collected
1300 # unclosed, which raises a spurious ``ResourceWarning`` in whatever
1301 # code happens to be running when the collection occurs.
1302 if self.db is not None:
1303 self.db.close()
1304 self.db = None
1305 atexit.unregister(self.stop)
1307 def stop(self) -> None:
1308 """This can be called from the main thread to safely stop this thread.
1310 Note that it does not attempt to write out remaining history before
1311 exiting. That should be done by calling the HistoryManager's
1312 end_session method."""
1313 if self._stopped:
1314 return
1315 self._stop_now = True
1317 self.save_flag.set()
1318 self._stopped = True
1319 if self.ident is not None and self != threading.current_thread():
1320 self.join()
1322 def __del__(self) -> None:
1323 self.stop()
1326# To match, e.g. ~5/8-~2/3, or ~4 (without trailing slash for full session)
1327# Session numbers: ~N or N/
1328# Line numbers: N (just digits, no ~)
1329# Range syntax: 4-6 (with end) or 4- (without end, means "onward")
1330range_re = re.compile(
1331 r"""
1332((?P<startsess>(?:~?\d+/)))?
1333(?P<start>\d+)?
1334((?P<sep>[\-:])
1335 ((?P<endsess>(?:~?\d+/)))?
1336 (?P<end>\d*))?
1337$""",
1338 re.VERBOSE,
1339)
1342def extract_hist_ranges(ranges_str: str) -> Iterable[tuple[int, int, int | None]]:
1343 """Turn a string of history ranges into 3-tuples of (session, start, stop).
1345 Empty string results in a `[(0, 1, None)]`, i.e. "everything from current
1346 session".
1348 Examples
1349 --------
1350 >>> list(extract_hist_ranges("~8/5-~7/4 2"))
1351 [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
1352 >>> list(extract_hist_ranges("~4/"))
1353 [(-4, 1, None)]
1354 >>> list(extract_hist_ranges("4-"))
1355 [(0, 4, None)]
1356 >>> list(extract_hist_ranges("~4/4-"))
1357 [(-4, 4, None)]
1358 """
1359 if ranges_str == "":
1360 yield (0, 1, None) # Everything from current session
1361 return
1363 for range_str in ranges_str.split():
1364 rmatch = range_re.match(range_str)
1365 if not rmatch:
1366 continue
1367 start = rmatch.group("start")
1368 sep = rmatch.group("sep")
1369 if start:
1370 start = int(start)
1371 end = rmatch.group("end")
1372 if sep == "-":
1373 end = (int(end) + 1) if end else None
1374 else:
1375 end = int(end) if end else start + 1
1376 else:
1377 if not rmatch.group("startsess"):
1378 continue
1379 start = 1
1380 end = None
1381 startsess = rmatch.group("startsess") or "0"
1382 endsess = rmatch.group("endsess") or startsess
1383 startsess = startsess.rstrip("/")
1384 endsess = endsess.rstrip("/")
1385 startsess = int(startsess.replace("~", "-"))
1386 endsess = int(endsess.replace("~", "-"))
1387 assert endsess >= startsess, "start session must be earlier than end session"
1389 if endsess == startsess:
1390 yield (startsess, start, end)
1391 continue
1392 # Multiple sessions in one range:
1393 yield (startsess, start, None)
1394 for sess in range(startsess + 1, endsess):
1395 yield (sess, 1, None)
1396 yield (endsess, 1, end)
1399def _format_lineno(session: int, line: int) -> str:
1400 """Helper function to format line numbers properly."""
1401 if session == 0:
1402 return str(line)
1403 return "{}#{}".format(session, line)