122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
from dataclasses import asdict
|
|
from datetime import UTC, datetime
|
|
from time import monotonic
|
|
from typing import Literal
|
|
from uuid import uuid4
|
|
|
|
from ipython_shell.utils import generate_call_id
|
|
|
|
from .models import AppInfo, CallOptions, CallResponse, ShellInfo
|
|
from .registry import ShellRegistry
|
|
from .shell import Shell
|
|
|
|
|
|
class App:
|
|
"""Coordinate named shells and expose frontend-shaped call responses."""
|
|
|
|
def __init__(self):
|
|
self.app_id = uuid4().hex
|
|
self.created_at = datetime.now(UTC).isoformat()
|
|
self._started_at = monotonic()
|
|
self._total_calls = 0
|
|
self._shell_registry = ShellRegistry(Shell)
|
|
self._issued_call_ids: set[str] = set()
|
|
|
|
@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]:
|
|
return self._shell_registry.select(shell_name)
|
|
|
|
def list_shells(self) -> list[ShellInfo]:
|
|
"""Return frontend-safe metadata for all known shells."""
|
|
return self._shell_registry.list_shells()
|
|
|
|
def create_shell(self) -> ShellInfo:
|
|
"""Create a fresh named shell and return its metadata."""
|
|
return self._shell_registry.create_shell()
|
|
|
|
def get_shell(self, shell_name: str) -> ShellInfo:
|
|
"""Return frontend-safe metadata for a single shell."""
|
|
return self._shell_registry.get_shell(shell_name)
|
|
|
|
def get_last_shell(self) -> ShellInfo:
|
|
"""Return metadata for the most recently selected 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."""
|
|
return AppInfo(
|
|
app_id=self.app_id,
|
|
created_at=self.created_at,
|
|
uptime_seconds=monotonic() - self._started_at,
|
|
shell_count=len(self.shells),
|
|
last_shell=self.last_shell,
|
|
total_calls=self._total_calls,
|
|
)
|
|
|
|
def get_shell_2(self, shell_name: Literal["last"] | str) -> ShellInfo:
|
|
"""Return frontend-safe metadata for a single shell."""
|
|
if shell_name == "last":
|
|
return self.get_last_shell()
|
|
return self.get_shell(shell_name)
|
|
|
|
def run_code(
|
|
self,
|
|
code: str,
|
|
shell_name: Literal["last", "new"] | str = "last",
|
|
*,
|
|
options: CallOptions | None = None,
|
|
) -> CallResponse:
|
|
"""Run code in the selected shell and return the complete call payload."""
|
|
if options is None:
|
|
options = CallOptions()
|
|
selected_name, shell = self._select_shell(shell_name)
|
|
call_id = self._new_call_id()
|
|
self._total_calls += 1
|
|
execution, events = shell.run_cell(code, call_id=call_id)
|
|
return CallResponse(
|
|
call_id=call_id,
|
|
shell_name=selected_name,
|
|
code=code,
|
|
options=options,
|
|
result=execution.to_response(),
|
|
events=events,
|
|
shell=shell.describe(),
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
"""Show the payload a frontend would receive from two related calls."""
|
|
app = App()
|
|
# Keep the transport clean: stdout is represented by an event in the
|
|
# payload, not emitted beside the JSON document.
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
calls = [
|
|
app.run_code("print('hello from the shell'); 2 + 2", shell_name="new"),
|
|
app.run_code(
|
|
"40 + 2",
|
|
shell_name="last",
|
|
options=CallOptions(collapsed=True),
|
|
),
|
|
]
|
|
print(json.dumps([asdict(call) for call in calls], indent=2, default=repr))
|