80 lines
2.0 KiB
Python
80 lines
2.0 KiB
Python
import traceback as traceback_module
|
|
|
|
from IPython.core.history import HistoryOutput
|
|
from IPython.core.interactiveshell import ExecutionResult
|
|
|
|
from .models import CallError, CallEvent
|
|
|
|
|
|
def event_from_history_output(
|
|
output: HistoryOutput,
|
|
*,
|
|
call_id: str,
|
|
sequence: int,
|
|
) -> CallEvent | None:
|
|
"""Convert one IPython ``HistoryOutput`` into the event model."""
|
|
if output.output_type == "out_stream":
|
|
return CallEvent(
|
|
call_id=call_id,
|
|
sequence=sequence,
|
|
kind="stdout",
|
|
text="".join(output.bundle.get("stream", [])),
|
|
)
|
|
|
|
if output.output_type == "err_stream":
|
|
return CallEvent(
|
|
call_id=call_id,
|
|
sequence=sequence,
|
|
kind="stderr",
|
|
text="".join(output.bundle.get("stream", [])),
|
|
)
|
|
|
|
if output.output_type in {"display_data", "execute_result"}:
|
|
return CallEvent(
|
|
call_id=call_id,
|
|
sequence=sequence,
|
|
kind=output.output_type,
|
|
data=dict(output.bundle),
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def error_from_execution_result(
|
|
execution: ExecutionResult,
|
|
) -> CallError | None:
|
|
"""Convert an IPython execution error into the application model."""
|
|
error = execution.error_before_exec or execution.error_in_exec
|
|
if error is None:
|
|
return None
|
|
|
|
return CallError(
|
|
ename=type(error).__name__,
|
|
evalue=str(error),
|
|
traceback=traceback_module.format_exception(error),
|
|
)
|
|
|
|
|
|
def event_from_execution_result(
|
|
execution: ExecutionResult,
|
|
*,
|
|
call_id: str,
|
|
sequence: int,
|
|
) -> CallEvent | None:
|
|
"""Convert an IPython execution error into an ``error`` event."""
|
|
error = error_from_execution_result(execution)
|
|
if error is None:
|
|
return None
|
|
|
|
return CallEvent(
|
|
call_id=call_id,
|
|
sequence=sequence,
|
|
kind="error",
|
|
text=error.evalue,
|
|
data={
|
|
"ename": error.ename,
|
|
"evalue": error.evalue,
|
|
"traceback": error.traceback,
|
|
},
|
|
)
|