diff --git a/IPython/core/history.py b/IPython/core/history.py index 0dd2abbb02b..791bcf58f5b 100644 --- a/IPython/core/history.py +++ b/IPython/core/history.py @@ -1,5 +1,7 @@ """ History related magics and functionality """ +from __future__ import annotations + # Copyright (c) IPython Development Team. # Distributed under the terms of the Modified BSD License. @@ -7,14 +9,6 @@ import atexit import datetime import re -try: - from sqlite3 import DatabaseError, OperationalError - import sqlite3 - sqlite3_found = True -except ModuleNotFoundError: - sqlite3_found = False - class DatabaseError(Exception):pass #type: ignore [no-redef] - class OperationalError(Exception):pass #type: ignore [no-redef] @@ -39,8 +33,20 @@ class OperationalError(Exception):pass #type: ignore [no-redef] from IPython.paths import locate_profile from IPython.utils.decorators import undoc -from typing import Iterable, Tuple, Optional +from typing import Iterable, Tuple, Optional, TYPE_CHECKING import typing +if TYPE_CHECKING: + from IPython.core.interactiveshell import InteractiveShell + from IPython.config.Configuration import Configuration + +try: + from sqlite3 import DatabaseError, OperationalError + import sqlite3 + sqlite3_found = True +except ModuleNotFoundError: + sqlite3_found = False + class DatabaseError(Exception):pass #type: ignore [no-redef] + class OperationalError(Exception):pass #type: ignore [no-redef] InOrInOut = typing.Union[str, Tuple[str, Optional[str]]] @@ -192,14 +198,14 @@ class HistoryAccessor(HistoryAccessorBase): ).tag(config=True) @default("connection_options") - def _default_connection_options(self): + def _default_connection_options(self) -> typing.Dict[str, bool]: return dict(check_same_thread=False) # The SQLite database db = Any() @observe('db') @only_when_enabled - def _db_changed(self, change): + def _db_changed(self, change): # type: ignore [no-untyped-def] """validate the db, since it can be an Instance of two different types""" new = change['new'] connection_types = (DummyDB, sqlite3.Connection) @@ -208,7 +214,7 @@ def _db_changed(self, change): (self.__class__.__name__, new) raise TraitError(msg) - def __init__(self, profile="default", hist_file="", **traits): + def __init__(self, profile:str="default", hist_file:str="", **traits:typing.Any) -> None: """Create a new history accessor. Parameters @@ -236,7 +242,7 @@ def __init__(self, profile="default", hist_file="", **traits): self.init_db() - def _get_hist_file_name(self, profile='default'): + def _get_hist_file_name(self, profile:str='default') -> Path: """Find the history file for the given profile name. This is overridden by the HistoryManager subclass, to use the shell's @@ -250,7 +256,7 @@ def _get_hist_file_name(self, profile='default'): return Path(locate_profile(profile)) / "history.sqlite" @catch_corrupt_db - def init_db(self): + def init_db(self) -> None: """Connect to the database, and create tables if necessary.""" if not self.enabled: self.db = DummyDB() @@ -259,7 +265,7 @@ def init_db(self): # use detect_types so that timestamps return datetime objects kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES) kwargs.update(self.connection_options) - self.db = sqlite3.connect(str(self.hist_file), **kwargs) + self.db = sqlite3.connect(str(self.hist_file), **kwargs) # type: ignore [call-overload] with self.db: self.db.execute( """CREATE TABLE IF NOT EXISTS sessions (session integer @@ -281,7 +287,7 @@ def init_db(self): # success! reset corrupt db count self._corrupt_db_counter = 0 - def writeout_cache(self): + def writeout_cache(self) -> None: """Overridden by HistoryManager to dump the cache before certain database lookups.""" pass @@ -289,7 +295,7 @@ def writeout_cache(self): ## ------------------------------- ## Methods for retrieving history: ## ------------------------------- - def _run_sql(self, sql, params, raw=True, output=False, latest=False): + def _run_sql(self, sql:str, params:typing.Tuple, raw:bool=True, output:bool=False, latest:bool=False) -> Iterable[Tuple[int, int, InOrInOut]]: """Prepares and runs an SQL query for the history database. Parameters @@ -324,7 +330,7 @@ def _run_sql(self, sql, params, raw=True, output=False, latest=False): @only_when_enabled @catch_corrupt_db - def get_session_info(self, session) -> Tuple[int, datetime.datetime, Optional[datetime.datetime], Optional[int], str]: + def get_session_info(self, session:int) -> Tuple[int, datetime.datetime, Optional[datetime.datetime], Optional[int], str]: """Get info about a session. Parameters @@ -349,7 +355,7 @@ def get_session_info(self, session) -> Tuple[int, datetime.datetime, Optional[da return self.db.execute(query, (session,)).fetchone() @catch_corrupt_db - def get_last_session_id(self): + def get_last_session_id(self) -> Optional[int]: """Get the last session ID currently in the database. Within IPython, this should be the same as the value stored in @@ -357,9 +363,10 @@ def get_last_session_id(self): """ for record in self.get_tail(n=1, include_latest=True): return record[0] + return None @catch_corrupt_db - def get_tail(self, n=10, raw=True, output=False, include_latest=False) -> Iterable[Tuple[int, int, str]]: + def get_tail(self, n:int=10, raw:bool=True, output:bool=False, include_latest:bool=False) -> Iterable[Tuple[int, int, InOrInOut]]: """Get the last n lines from the history database. Parameters @@ -388,8 +395,7 @@ def get_tail(self, n=10, raw=True, output=False, include_latest=False) -> Iterab return reversed(list(cur)) @catch_corrupt_db - def search(self, pattern="*", raw=True, search_raw=True, - output=False, n=None, unique=False): + def search(self, pattern:str="*", raw:bool=True, search_raw:bool=True, output:bool=False, n:Optional[int]=None, unique:bool=False) -> Iterable[Tuple[int, int, InOrInOut]]: """Search the database using unix glob-style matching (wildcards * and ?). @@ -416,7 +422,7 @@ def search(self, pattern="*", raw=True, search_raw=True, tosearch = "history." + tosearch self.writeout_cache() sqlform = "WHERE %s GLOB ?" % tosearch - params = (pattern,) + params:typing.Tuple[typing.Any, ...] = (pattern,) if unique: sqlform += ' GROUP BY {0}'.format(tosearch) if n is not None: @@ -430,7 +436,7 @@ def search(self, pattern="*", raw=True, search_raw=True, return cur @catch_corrupt_db - def get_range(self, session, start=1, stop=None, raw=True,output=False): + def get_range(self, session:int, start:int=1, stop:Optional[int]=None, raw:bool=True, output:bool=False) -> Iterable[Tuple[int, int, InOrInOut]]: """Retrieve input by session. Parameters @@ -457,6 +463,7 @@ def get_range(self, session, start=1, stop=None, raw=True,output=False): (session, line, input) if output is False, or (session, line, (input, output)) if output is True. """ + params:typing.Tuple[typing.Any, ...] if stop: lineclause = "line >= ? AND line < ?" params = (session, start, stop) @@ -467,7 +474,7 @@ def get_range(self, session, start=1, stop=None, raw=True,output=False): return self._run_sql("WHERE session==? AND %s" % lineclause, params, raw=raw, output=output) - def get_range_by_str(self, rangestr, raw=True, output=False): + def get_range_by_str(self, rangestr:str, raw:bool=True, output:bool=False) -> Iterable[Tuple[int, int, InOrInOut]]: """Get lines of history from a string of ranges, as used by magic commands %hist, %save, %macro, etc. @@ -498,7 +505,7 @@ class HistoryManager(HistoryAccessor): # An instance of the IPython shell we are attached to shell = Instance('IPython.core.interactiveshell.InteractiveShellABC', - allow_none=True) + allow_none=False) # Lists to hold processed and raw history. These start with a blank entry # so that we can index them starting from 1 input_hist_parsed = List([""]) @@ -507,7 +514,7 @@ class HistoryManager(HistoryAccessor): dir_hist: List = List() @default("dir_hist") - def _dir_hist_default(self): + def _dir_hist_default(self) -> typing.List[Path]: try: return [Path.cwd()] except OSError: @@ -517,10 +524,10 @@ def _dir_hist_default(self): # execution count. output_hist = Dict() # The text/plain repr of outputs. - output_hist_reprs: typing.Dict[int, str] = Dict() + output_hist_reprs: typing.Dict[int, str] = Dict() # type: ignore [assignment] # The number of the current session in the history database - session_number:int = Integer() + session_number:int = Integer() #type: ignore [assignment] db_log_output = Bool(False, help="Should the history database include output? (default: no)" @@ -552,7 +559,7 @@ def _dir_hist_default(self): # an exit call). _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$") - def __init__(self, shell=None, config=None, **traits): + def __init__(self, shell: InteractiveShell, config: Optional[Configuration] = None, **traits: typing.Any): """Create a new history manager associated with a shell instance. """ super().__init__(shell=shell, config=config, @@ -579,7 +586,7 @@ def __init__(self, shell=None, config=None, **traits): ) self.hist_file = ":memory:" - def _get_hist_file_name(self, profile:str=None): + def _get_hist_file_name(self, profile:Optional[str]=None) -> Path: """Get default history file name based on the Shell's profile. The profile parameter is ignored, but must exist for compatibility with