shared shell registry

This commit is contained in:
lda
2026-08-30 22:49:34 +07:00 Verified
parent 0b87620852
commit 6a8546cb0f
8 changed files with 262 additions and 68 deletions
+27 -33
View File
@@ -7,9 +7,10 @@ from time import monotonic
from typing import Literal
from uuid import uuid4
from ipython_shell.utils import generate_good_names
from ipython_shell.utils import generate_call_id
from .models import AppInfo, CallOptions, CallResponse, ShellInfo
from .registry import ShellRegistry
from .shell import Shell
@@ -21,52 +22,45 @@ class App:
self.created_at = datetime.now(UTC).isoformat()
self._started_at = monotonic()
self._total_calls = 0
self.shells: dict[str, Shell] = {}
self.last_shell: str | None = None
self._shell_registry = ShellRegistry(Shell)
self._issued_call_ids: set[str] = set()
def _new_shell(self) -> tuple[str, Shell]:
name = generate_good_names()
while name in self.shells:
name = generate_good_names()
shell = Shell(name)
self.shells[name] = shell
self.last_shell = name
return name, shell
@property
def shells(self) -> dict[str, Shell]:
"""Expose the managed shells for compatibility with the old API."""
return self._shell_registry.shells
@property
def last_shell(self) -> str | None:
"""Return the name of the most recently selected shell."""
return self._shell_registry.last_shell
def _select_shell(self, shell_name: str) -> tuple[str, Shell]:
if shell_name == "new":
return self._new_shell()
if shell_name == "last":
if self.last_shell is None:
return self._new_shell()
return self.last_shell, self.shells[self.last_shell]
if shell_name not in self.shells:
self.shells[shell_name] = Shell(shell_name)
self.last_shell = shell_name
return shell_name, self.shells[shell_name]
return self._shell_registry.select(shell_name)
def list_shells(self) -> list[ShellInfo]:
"""Return frontend-safe metadata for all known shells."""
return [shell.describe() for shell in self.shells.values()]
return self._shell_registry.list_shells()
def create_shell(self) -> ShellInfo:
"""Create a fresh named shell and return its metadata."""
_, shell = self._new_shell()
return shell.describe()
return self._shell_registry.create_shell()
def get_shell(self, shell_name: str) -> ShellInfo:
"""Return frontend-safe metadata for a single shell."""
if shell_name not in self.shells:
raise KeyError(f"Shell {shell_name} does not exist")
return self.shells[shell_name].describe()
return self._shell_registry.get_shell(shell_name)
def get_last_shell(self) -> ShellInfo:
"""Return metadata for the most recently selected shell."""
if self.last_shell is None:
raise KeyError("No shell exists yet")
return self.get_shell(self.last_shell)
return self._shell_registry.get_last_shell()
def _new_call_id(self) -> str:
"""Generate a readable call ID that is unique for this app."""
call_id = generate_call_id()
while call_id in self._issued_call_ids:
call_id = generate_call_id()
self._issued_call_ids.add(call_id)
return call_id
def info(self) -> AppInfo:
"""Return runtime metadata without initializing or inspecting shells."""
@@ -96,7 +90,7 @@ class App:
if options is None:
options = CallOptions()
selected_name, shell = self._select_shell(shell_name)
call_id = uuid4().hex
call_id = self._new_call_id()
self._total_calls += 1
execution, events = shell.run_cell(code, call_id=call_id)
return CallResponse(