Skip to content

Commit

Permalink
more types
Browse files Browse the repository at this point in the history
  • Loading branch information
Carreau committed Dec 12, 2024
1 parent 383cb4f commit d70585a
Showing 1 changed file with 20 additions and 20 deletions.
40 changes: 20 additions & 20 deletions IPython/core/history.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ 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, TypeVar
from typing import Iterable, Tuple, Optional
import typing

InOrInOut = typing.Union[str, Tuple[str, Optional[str]]]
Expand All @@ -53,21 +53,21 @@ class DummyDB:
"""Dummy DB that will act as a black hole for history.
Only used in the absence of sqlite"""
def execute(*args, **kwargs):
def execute(*args:typing.Any, **kwargs:typing.Any) -> typing.List:
return []

def commit(self, *args, **kwargs):
def commit(self, *args, **kwargs): # type: ignore [no-untyped-def]
pass

def __enter__(self, *args, **kwargs):
def __enter__(self, *args, **kwargs): # type: ignore [no-untyped-def]
pass

def __exit__(self, *args, **kwargs):
def __exit__(self, *args, **kwargs): # type: ignore [no-untyped-def]
pass


@decorator
def only_when_enabled(f, self, *a, **kw):
def only_when_enabled(f, self, *a, **kw): # type: ignore [no-untyped-def]
"""Decorator: return an empty list in the absence of sqlite."""
if not self.enabled:
return []
Expand All @@ -80,7 +80,7 @@ def only_when_enabled(f, self, *a, **kw):
_SAVE_DB_SIZE = 16384

@decorator
def catch_corrupt_db(f, self, *a, **kw):
def catch_corrupt_db(f, self, *a, **kw): # type: ignore [no-untyped-def]
"""A decorator which wraps HistoryAccessor method calls to catch errors from
a corrupt SQLite database, move the old database out of the way, and create
a new one.
Expand Down Expand Up @@ -128,17 +128,16 @@ def catch_corrupt_db(f, self, *a, **kw):
class HistoryAccessorBase(LoggingConfigurable):
"""An abstract class for History Accessors """

def get_tail(self, n:int=10, raw:bool=True, output:bool=False, include_latest:bool=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]]:
raise NotImplementedError

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]]:
raise NotImplementedError

def get_range(self, session: int, start: int = 1, stop: Optional[int] = None, raw: bool = True, output: bool = False) -> Iterable[Tuple[int, int, InOrInOut]]:
raise NotImplementedError

def get_range_by_str(self, rangestr: str, raw: bool = True, output: bool = False):
def get_range_by_str(self, rangestr: str, raw: bool = True, output: bool = False)->Iterable[Tuple[int, int, InOrInOut]]:
raise NotImplementedError


Expand Down Expand Up @@ -325,7 +324,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):
def get_session_info(self, session) -> Tuple[int, datetime.datetime, Optional[datetime.datetime], Optional[int], str]:
"""Get info about a session.
Parameters
Expand Down Expand Up @@ -518,10 +517,10 @@ def _dir_hist_default(self):
# execution count.
output_hist = Dict()
# The text/plain repr of outputs.
output_hist_reprs: Dict[int, str] = Dict()
output_hist_reprs: typing.Dict[int, str] = Dict()

# The number of the current session in the history database
session_number = Integer()
session_number:int = Integer()

db_log_output = Bool(False,
help="Should the history database include output? (default: no)"
Expand Down Expand Up @@ -580,7 +579,7 @@ def __init__(self, shell=None, config=None, **traits):
)
self.hist_file = ":memory:"

def _get_hist_file_name(self, profile=None):
def _get_hist_file_name(self, profile:str=None):
"""Get default history file name based on the Shell's profile.
The profile parameter is ignored, but must exist for compatibility with
Expand All @@ -589,7 +588,7 @@ def _get_hist_file_name(self, profile=None):
return Path(profile_dir) / "history.sqlite"

@only_when_enabled
def new_session(self, conn=None):
def new_session(self, conn:Optional[sqlite3.Connection]=None) -> None:
"""Get a new session number."""
if conn is None:
conn = self.db
Expand All @@ -600,9 +599,10 @@ def new_session(self, conn=None):
NULL, '') """,
(datetime.datetime.now().isoformat(" "),),
)
assert isinstance(cur.lastrowid, int)
self.session_number = cur.lastrowid

def end_session(self):
def end_session(self) -> None:
"""Close the database session, filling in the end time and line count."""
self.writeout_cache()
with self.db:
Expand All @@ -617,13 +617,13 @@ def end_session(self):
)
self.session_number = 0

def name_session(self, name):
def name_session(self, name:str) -> None:
"""Give the current session a name in the history database."""
with self.db:
self.db.execute("UPDATE sessions SET remark=? WHERE session==?",
(name, self.session_number))

def reset(self, new_session=True):
def reset(self, new_session:bool=True)-> None :
"""Clear the session history, releasing all object references, and
optionally open a new session."""
self.output_hist.clear()
Expand All @@ -640,7 +640,7 @@ def reset(self, new_session=True):
# ------------------------------
# Methods for retrieving history
# ------------------------------
def get_session_info(self, session=0):
def get_session_info(self, session:int=0) -> Tuple[int, datetime.datetime, Optional[datetime.datetime], Optional[int], str]:
"""Get info about a session.
Parameters
Expand Down

0 comments on commit d70585a

Please sign in to comment.