lda/ipython-demo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

carefully setup shell + rename

Browse Source
This commit is contained in:
lda
2026-08-30 10:25:06 +07:00 Verified
parent 16553b82c7
commit ad716bfb7f
15 changed files with 368 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ipython_demo/shell.py
-194
View File
@@ -1,194 +0,0 @@
from collections import defaultdict
from datetime import UTC, datetime
from uuid import uuid4
from IPython.core.interactiveshell import InteractiveShell
from IPython.core.pylabtools import (
activate_matplotlib,
import_pylab,
mpl_runner,
select_figure_formats,
)
from matplotlib_inline.backend_inline import configure_inline_support
from .events import (
error_from_execution_result,
event_from_execution_result,
event_from_history_output,
)
from .models import CallEvent, CallResult, ShellInfo, ShellStatus
class Shell:
"""Own one configured, in-process IPython shell."""
def __init__(self, name: str):
self.name = name
self.shell_id = uuid4().hex
self.created_at = datetime.now(UTC).isoformat()
self.last_used_at: str | None = None
self.status: ShellStatus = "ready"
self._last_execution_count = 0
self._shell: InteractiveShell | None = None
@property
def shell(self) -> InteractiveShell:
if self._shell is not None:
return self._shell
return self.init_shell()
def init_shell(self) -> InteractiveShell:
if self._shell is not None:
return self._shell
self._shell = InteractiveShell()
setup_shell(self._shell, plt=True, pylab=True)
return self._shell
def run_cell(
self,
code: str,
*,
call_id: str,
) -> tuple[CallResult, list[CallEvent]]:
"""Execute one cell and return its result plus ordered output events."""
self.status = "running"
try:
call_result, events = run_cell_and_collect(
self.shell,
code,
call_id=call_id,
)
except Exception:
self.status = "error"
raise
else:
self.status = "ready"
self.last_used_at = datetime.now(UTC).isoformat()
self._last_execution_count = call_result.execution_count or 0
return call_result, events
def describe(self) -> ShellInfo:
"""Return metadata without initializing an unused shell."""
return ShellInfo(
shell_id=self.shell_id,
name=self.name,
status=self.status,
execution_count=self._last_execution_count,
created_at=self.created_at,
last_used_at=self.last_used_at,
capabilities=(
"stdout",
"stderr",
"display_data",
"execute_result",
"error",
"matplotlib-inline",
),
)
def isolate_output_history(shell: InteractiveShell) -> None:
"""Give this shell an instance-owned output ledger.
IPython declares ``HistoryManager.outputs`` on the class. Shadowing it
during shell setup prevents output records from leaking across shells.
"""
history_manager = shell.history_manager
if history_manager is None:
raise RuntimeError("InteractiveShell has no history manager")
history_manager.outputs = defaultdict(list)
def run_cell_and_collect(
shell: InteractiveShell,
code: str,
*,
call_id: str,
) -> tuple[CallResult, list[CallEvent]]:
"""Run one cell and snapshot its output records after execution."""
history_manager = shell.history_manager
if history_manager is None:
raise RuntimeError("InteractiveShell has no history manager")
if "outputs" not in history_manager.__dict__:
isolate_output_history(shell)
execution = shell.run_cell(code, store_history=True)
records = (
tuple(history_manager.outputs.get(execution.execution_count, ()))
if execution.execution_count is not None
else ()
)
events = [
event_from_history_output(
record,
call_id=call_id,
sequence=sequence,
)
for sequence, record in enumerate(records)
]
events = [event for event in events if event is not None]
error = error_from_execution_result(execution)
if error is not None:
error_event = event_from_execution_result(
execution,
call_id=call_id,
sequence=len(events),
)
if error_event is not None:
events.append(error_event)
return (
CallResult(
call_id=call_id,
execution_count=execution.execution_count,
result=execution.result,
error=error,
),
events,
)
def setup_shell(shell: InteractiveShell, plt: bool = False, pylab: bool = True):
"""Configure an ``InteractiveShell`` without requiring GUI support.
``InteractiveShell.enable_matplotlib`` and ``enable_pylab`` eventually
call ``enable_gui``. The base shell intentionally does not implement that
method, so the lower-level matplotlib-inline setup is used here instead.
"""
history_manager = shell.history_manager
if history_manager is not None:
history_manager.enabled = False
# ``HistoryManager.outputs`` is class-level in the IPython version used
# here. Give every managed shell an instance-local mapping so execution
# records cannot leak between Shell objects.
if "outputs" not in history_manager.__dict__:
isolate_output_history(shell)
backend = "module://matplotlib_inline.backend_inline"
# This is the non-GUI part of enable_matplotlib().
if plt:
activate_matplotlib(backend)
shell.magics_manager.registry["ExecutionMagics"].default_runner = mpl_runner(
shell.safe_execfile
)
# This connects matplotlib-inline to THIS shell.
configure_inline_support(shell, backend)
# Optional explicit format selection; configure_inline_support()
# also sets up the configured default formats.
select_figure_formats(
shell,
{"svg", "pdf", "png", "jpg"},
)
if pylab:
# Match enable_pylab's namespace behavior without its GUI activation:
# hidden names are used by %who/%whos bookkeeping as well as by code.
pylab_ns: dict[str, object] = {}
import_pylab(pylab_ns, import_all=False) # no import * allowed
shell.user_ns.update(pylab_ns)
shell.user_ns_hidden.update(pylab_ns)
src/ipython_mcp/app.py
+3 -3
View File
@@ -5,11 +5,11 @@ from typing import Annotated, Literal
from fastmcp import FastMCP
from pydantic import Field
import ipython_demo
from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo
import ipython_shell
from ipython_shell.models import AppInfo, CallOptions, CallResponse, ShellInfo
app = FastMCP(name="IPython Demo")
shell_app = ipython_demo.App()
shell_app = ipython_shell.App()
@app.tool(title="Get app info")
src/ipython_demo/__init__.py → src/ipython_shell/__init__.py
+2
View File
@@ -22,6 +22,7 @@ from .models import (
from .shell import (
Shell,
isolate_output_history,
carefully_setup_shell,
run_cell_and_collect,
setup_shell,
)
@@ -41,6 +42,7 @@ __all__ = [
"Shell",
"ShellInfo",
"ShellStatus",
"carefully_setup_shell",
"error_from_execution_result",
"event_from_execution_result",
"event_from_history_output",
src/ipython_demo/app.py → src/ipython_shell/app.py
+1 -1
View File
@@ -7,7 +7,7 @@ from time import monotonic
from typing import Literal
from uuid import uuid4
from ipython_demo.utils import generate_good_names
from ipython_shell.utils import generate_good_names
from .models import AppInfo, CallOptions, CallResponse, ShellInfo
from .shell import Shell
src/ipython_demo/events.py → src/ipython_shell/events.py
View File
src/ipython_demo/models.py → src/ipython_shell/models.py
View File
src/ipython_shell/shell.py
+318
View File
@@ -0,0 +1,318 @@
from collections import defaultdict
from datetime import UTC, datetime
from uuid import uuid4
from IPython.core.interactiveshell import InteractiveShell
from IPython.core.pylabtools import (
activate_matplotlib,
find_gui_and_backend,
import_pylab,
mpl_runner,
select_figure_formats,
)
from matplotlib_inline.backend_inline import configure_inline_support, flush_figures
from .events import (
error_from_execution_result,
event_from_execution_result,
event_from_history_output,
)
from .models import CallEvent, CallResult, ShellInfo, ShellStatus
class Shell:
"""Own one configured, in-process IPython shell."""
def __init__(self, name: str):
self.name = name
self.shell_id = uuid4().hex
self.created_at = datetime.now(UTC).isoformat()
self.last_used_at: str | None = None
self.status: ShellStatus = "ready"
self._last_execution_count = 0
self._shell: InteractiveShell | None = None
@property
def shell(self) -> InteractiveShell:
if self._shell is not None:
return self._shell
return self.init_shell()
@classmethod
def using_shell(cls, shell: InteractiveShell, name: str) -> Shell:
"""Create a Shell object that wraps an existing InteractiveShell.
The caller owns setup for this low-level adapter.
"""
obj = cls(name)
obj._shell = shell
return obj
@classmethod
def init_instance(cls, name: str = "default") -> Shell:
"""Create a Shell around IPython's process-wide instance.
``matplotlib-inline`` reaches the shell through IPython's global
``get_ipython()`` lookup. Wrapping ``InteractiveShell.instance()``
keeps those two identities aligned without changing the singleton's
existing configuration.
"""
return cls.using_shell(InteractiveShell.instance(), name)
def init_shell(self) -> InteractiveShell:
if self._shell is not None:
return self._shell
self._shell = InteractiveShell()
setup_shell(self._shell, plt=True, pylab=True)
return self._shell
def run_cell(
self,
code: str,
*,
call_id: str,
) -> tuple[CallResult, list[CallEvent]]:
"""Execute one cell and return its result plus ordered output events."""
self.status = "running"
try:
call_result, events = run_cell_and_collect(
self.shell,
code,
call_id=call_id,
)
except Exception:
self.status = "error"
raise
else:
self.status = "ready"
self.last_used_at = datetime.now(UTC).isoformat()
self._last_execution_count = call_result.execution_count or 0
return call_result, events
def describe(self) -> ShellInfo:
"""Return metadata without initializing an unused shell."""
return ShellInfo(
shell_id=self.shell_id,
name=self.name,
status=self.status,
execution_count=self._last_execution_count,
created_at=self.created_at,
last_used_at=self.last_used_at,
capabilities=(
"stdout",
"stderr",
"display_data",
"execute_result",
"error",
"matplotlib-inline",
),
)
def isolate_output_history(shell: InteractiveShell) -> None:
"""Give this shell an instance-owned output ledger.
IPython declares ``HistoryManager.outputs`` on the class. Shadowing it
during shell setup prevents output records from leaking across shells.
"""
history_manager = shell.history_manager
if history_manager is None:
raise RuntimeError("InteractiveShell has no history manager")
history_manager.outputs = defaultdict(list)
def run_cell_and_collect(
shell: InteractiveShell,
code: str,
*,
call_id: str,
) -> tuple[CallResult, list[CallEvent]]:
"""Run one cell and snapshot its output records after execution."""
history_manager = shell.history_manager
if history_manager is None:
raise RuntimeError("InteractiveShell has no history manager")
if "outputs" not in history_manager.__dict__:
isolate_output_history(shell)
execution = shell.run_cell(code, store_history=True)
records = (
tuple(history_manager.outputs.get(execution.execution_count, ()))
if execution.execution_count is not None
else ()
)
events = [
event_from_history_output(
record,
call_id=call_id,
sequence=sequence,
)
for sequence, record in enumerate(records)
]
events = [event for event in events if event is not None]
error = error_from_execution_result(execution)
if error is not None:
error_event = event_from_execution_result(
execution,
call_id=call_id,
sequence=len(events),
)
if error_event is not None:
events.append(error_event)
return (
CallResult(
call_id=call_id,
execution_count=execution.execution_count,
result=execution.result,
error=error,
),
events,
)
def setup_shell(
shell: InteractiveShell,
*,
isolate_output_hist: bool = True,
plt: bool = False,
pylab: bool = True,
):
"""Configure an ``InteractiveShell`` without requiring GUI support.
``InteractiveShell.enable_matplotlib`` and ``enable_pylab`` eventually
call ``enable_gui``. The base shell intentionally does not implement that
method, so the lower-level matplotlib-inline setup is used here instead.
"""
setup_shell_history_disabled(shell)
# ``HistoryManager.outputs`` is class-level in the IPython version used
# here. Give every managed shell an instance-local mapping so execution
# records cannot leak between Shell objects.
if (
isolate_output_hist
and shell.history_manager is not None
and "outputs" not in shell.history_manager.__dict__
):
isolate_output_history(shell)
carefully_setup_shell(
shell,
activate=plt,
pylab=pylab,
install_runner=plt,
)
def setup_shell_history_disabled(shell: InteractiveShell):
"""Disable the shell's history manager to mostly disable writing to disk."""
if shell.history_manager is not None:
shell.history_manager.enabled = False
def carefully_setup_shell(
shell: InteractiveShell,
*,
gui: str = "inline",
activate: bool = True,
formats: set[str] | None = None,
pylab: bool = False,
import_all: bool = False,
install_runner: bool = False,
) -> tuple[str | None, str, list[str]]:
"""Add only the requested Matplotlib and pylab support to one shell.
This deliberately does not call ``enable_gui``. It performs the useful
parts of ``InteractiveShell.enable_matplotlib`` and ``enable_pylab`` for
shells whose base class has no GUI implementation.
"""
import matplotlib
selected_gui, backend = find_gui_and_backend(
gui,
shell.pylab_gui_select,
)
wanted_formats = formats or {"svg", "png", "pdf", "jpg"}
# select_figure_formats() clears Figure printers, so capture them before
# configuring inline support and restore them with our requested formats.
existing_formats = figure_formats_for(shell)
if activate and matplotlib.get_backend().lower() != backend.lower():
activate_matplotlib(backend)
if not inline_support_present(shell, backend):
configure_inline_support(shell, backend)
select_figure_formats(shell, existing_formats | wanted_formats)
if install_runner:
execution_magics = shell.magics_manager.registry["ExecutionMagics"]
if execution_magics.default_runner is None:
execution_magics.default_runner = mpl_runner(shell.safe_execfile)
clobbered = (
enable_pylab_lite(shell, import_all=import_all) if pylab else []
)
return selected_gui, backend, clobbered
def inline_support_present(shell: InteractiveShell, backend: str) -> bool:
"""Return whether this shell has the requested inline flush support."""
callbacks = shell.events.callbacks.get("post_execute", [])
is_inline = backend.lower() in {
"inline",
"module://matplotlib_inline.backend_inline",
}
return (flush_figures in callbacks) if is_inline else (flush_figures not in callbacks)
def figure_formats_for(shell: InteractiveShell) -> set[str]:
"""Return MIME figure formats currently registered on one shell."""
from matplotlib.figure import Figure
mime_to_format = {
"image/png": "png",
"image/jpeg": "jpg",
"image/svg+xml": "svg",
"application/pdf": "pdf",
}
return {
format_name
for mime, format_name in mime_to_format.items()
if Figure in shell.display_formatter.formatters[mime].type_printers
}
def enable_pylab_lite(
shell: InteractiveShell,
*,
import_all: bool = True,
) -> list[str]:
"""Load pylab names without GUI activation and report overwritten names."""
pylab_ns: dict[str, object] = {}
import_pylab(pylab_ns, import_all)
ignored = {"__builtins__"}
candidates = set(pylab_ns).intersection(shell.user_ns).difference(ignored)
clobbered = [
name
for name in candidates
if shell.user_ns[name] is not pylab_ns[name]
]
shell.user_ns.update(pylab_ns)
shell.user_ns_hidden.update(pylab_ns)
return clobbered
_default: Shell | None = None
def get_default_shell() -> Shell:
"""Return the lazily-created, configured process-wide shell."""
global _default
if _default is None:
_default = Shell.init_instance("default")
return _default
src/ipython_demo/utils.py → src/ipython_shell/utils.py
View File
src/ipython_webapp/app.py
+2 -2
View File
@@ -2,8 +2,8 @@ from typing import Annotated, Literal
from fastapi import Body, Depends, FastAPI, HTTPException, Path, Query
from ipython_demo import App
from ipython_demo.models import CallOptions
from ipython_shell import App
from ipython_shell.models import CallOptions
app = FastAPI(title="IPython Demo", version="0.1.0")
src/langchain_demo/ipython_wrapper.py
+2 -2
View File
@@ -2,8 +2,8 @@ from typing import Literal
from langchain.tools import tool
from ipython_demo import App
from ipython_demo.models import AppInfo, CallOptions, CallResponse, ShellInfo
from ipython_shell import App
from ipython_shell.models import AppInfo, CallOptions, CallResponse, ShellInfo
app = App() # global, again
tests/test_app.py
+3 -3
View File
@@ -4,9 +4,9 @@ import re
import unittest
from unittest.mock import Mock, patch
from ipython_demo import utils
from ipython_demo.app import App
from ipython_demo.models import CallOptions, CallResponse
from ipython_shell import utils
from ipython_shell.app import App
from ipython_shell.models import CallOptions, CallResponse
class AppTests(unittest.TestCase):
tests/test_info.py
+1 -1
View File
@@ -2,7 +2,7 @@ import unittest
from fastapi.testclient import TestClient
from ipython_demo import App, AppInfo
from ipython_shell import App, AppInfo
from ipython_mcp.app import app as mcp_app
from ipython_webapp.app import app as web_app
from langchain_demo.ipython_wrapper import get_app_info
tests/test_my_ideas.py
+1 -1
View File
@@ -4,7 +4,7 @@ from typing import Any
from IPython.core.interactiveshell import InteractiveShell
from ipython_demo.shell import Shell
from ipython_shell.shell import Shell
class TestMyIdeas(unittest.TestCase):
tests/test_serialization.py
+1 -1
View File
@@ -5,7 +5,7 @@ import unittest
from fastapi.encoders import jsonable_encoder
from fastapi.testclient import TestClient
from ipython_demo import App, Shell
from ipython_shell import App, Shell
from ipython_webapp.app import app as web_app
tests/test_shell.py
+34 -3
View File
@@ -8,9 +8,9 @@ from IPython.core.history import HistoryOutput
from IPython.core.interactiveshell import InteractiveShell
from IPython.utils import capture
import ipython_demo.events as events_module
import ipython_demo.models as models_module
import ipython_demo.shell as shell_module
import ipython_shell.events as events_module
import ipython_shell.models as models_module
import ipython_shell.shell as shell_module
PACKAGE_ROOT = Path(__file__).parents[1] / "src" / "ipython_demo"
@@ -123,6 +123,37 @@ class ShellTests(unittest.TestCase):
shell.magics_manager.registry["ExecutionMagics"].default_runner
)
def test_carefully_setup_shell_preserves_formats_and_is_idempotent(self):
from matplotlib.figure import Figure
from matplotlib_inline.backend_inline import flush_figures
shell = InteractiveShell()
shell_module.carefully_setup_shell(shell, formats={"svg"})
shell_module.carefully_setup_shell(shell, formats={"png"})
callbacks = shell.events.callbacks["post_execute"]
self.assertEqual(callbacks.count(flush_figures), 1)
self.assertIn(
Figure,
shell.display_formatter.formatters["image/svg+xml"].type_printers,
)
self.assertIn(
Figure,
shell.display_formatter.formatters["image/png"].type_printers,
)
def test_enable_pylab_lite_reports_clobbered_names(self):
shell = InteractiveShell()
original_np = object()
shell.user_ns["np"] = original_np
clobbered = shell_module.enable_pylab_lite(shell, import_all=False)
self.assertIn("np", clobbered)
self.assertIsNot(shell.user_ns["np"], original_np)
self.assertIs(shell.user_ns["np"], shell.user_ns_hidden["np"])
def test_run_cell_does_not_use_global_capture_output(self):
wrapper = shell_module.Shell("test")
wrapper.init_shell()