##// END OF EJS Templates
Use check_same_thread=False by default for history sqlite db (#13886)...
Use check_same_thread=False by default for history sqlite db (#13886) I had run into the following exception while trying to use IPython in a thread: ``` Traceback (most recent call last): File "/usr/local/lib/python3.11/dist-packages/IPython/core/interactiveshell.py", line 3745, in atexit_operations self._atexit_once() File "/usr/local/lib/python3.11/dist-packages/IPython/core/interactiveshell.py", line 3728, in _atexit_once self.history_manager.end_session() File "/usr/local/lib/python3.11/dist-packages/IPython/core/history.py", line 576, in end_session self.writeout_cache() File "/usr/local/lib/python3.11/dist-packages/decorator.py", line 232, in fun return caller(func, *(extras + args), **kw) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/dist-packages/IPython/core/history.py", line 60, in only_when_enabled return f(self, *a, **kw) ^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/dist-packages/IPython/core/history.py", line 831, in writeout_cache self._writeout_input_cache(conn) File "/usr/local/lib/python3.11/dist-packages/IPython/core/history.py", line 812, in _writeout_input_cache with conn: sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 139673788811008 and this is thread id 139673823184704. ``` And discovered that an issue (#680) has been open for it since 2011. Back in [2012](https://github.com/ipython/ipython/issues/680#issuecomment-3444922), it seems like the only reason not to fix it was that the parameter `check_same_thread` was not documented, but now [it is](https://docs.python.org/3.11/library/sqlite3.html#sqlite3.connect), and it has been at least since [3.6](https://docs.python.org/3.6/library/sqlite3.html#sqlite3.connect). Note that according to the docs: > check_same_thread ([bool](https://docs.python.org/3/library/functions.html#bool)) – If True (default), only the creating thread may use the connection. If False, the connection may be shared across multiple threads; if so, write operations should be serialized by the user to avoid data corruption. But I don't think this is an issue here. The operations should be synchronized by the user only *on the same connection object*, and if I'm not mistaken, if two instances of IPython were started in separate threads, they'd each have their own history manager with its own connection. The exception above (and the one in the related issue) is raised from `atexit`, when the main thread is running finalizations registered by other threads. At that point, the operations are serial anyway because they are executed in a single thread.

File last commit:

r27386:82d1a374
r28320:2da6fb98 merge
Show More
conftest.py
87 lines | 2.6 KiB | text/x-python | PythonLexer
import builtins
import inspect
import os
import pathlib
import shutil
import sys
import types
import pytest
# Must register before it gets imported
pytest.register_assert_rewrite("IPython.testing.tools")
from .testing import tools
def pytest_collection_modifyitems(items):
"""This function is automatically run by pytest passing all collected test
functions.
We use it to add asyncio marker to all async tests and assert we don't use
test functions that are async generators which wouldn't make sense.
"""
for item in items:
if inspect.iscoroutinefunction(item.obj):
item.add_marker("asyncio")
assert not inspect.isasyncgenfunction(item.obj)
def get_ipython():
from .terminal.interactiveshell import TerminalInteractiveShell
if TerminalInteractiveShell._instance:
return TerminalInteractiveShell.instance()
config = tools.default_config()
config.TerminalInteractiveShell.simple_prompt = True
# Create and initialize our test-friendly IPython instance.
shell = TerminalInteractiveShell.instance(config=config)
return shell
@pytest.fixture(scope='session', autouse=True)
def work_path():
path = pathlib.Path("./tmp-ipython-pytest-profiledir")
os.environ["IPYTHONDIR"] = str(path.absolute())
if path.exists():
raise ValueError('IPython dir temporary path already exists ! Did previous test run exit successfully ?')
path.mkdir()
yield
shutil.rmtree(str(path.resolve()))
def nopage(strng, start=0, screen_lines=0, pager_cmd=None):
if isinstance(strng, dict):
strng = strng.get("text/plain", "")
print(strng)
def xsys(self, cmd):
"""Replace the default system call with a capturing one for doctest.
"""
# We use getoutput, but we need to strip it because pexpect captures
# the trailing newline differently from commands.getoutput
print(self.getoutput(cmd, split=False, depth=1).rstrip(), end="", file=sys.stdout)
sys.stdout.flush()
# for things to work correctly we would need this as a session fixture;
# unfortunately this will fail on some test that get executed as _collection_
# time (before the fixture run), in particular parametrized test that contain
# yields. so for now execute at import time.
#@pytest.fixture(autouse=True, scope='session')
def inject():
builtins.get_ipython = get_ipython
builtins._ip = get_ipython()
builtins.ip = get_ipython()
builtins.ip.system = types.MethodType(xsys, ip)
builtins.ip.builtin_trap.activate()
from .core import page
page.pager_page = nopage
# yield
inject()