JezK

Edit File: console.py

import code
import sys
import typing as t
from contextvars import ContextVar
from types import CodeType

from markupsafe import escape

from .repr import debug_repr
from .repr import dump
from .repr import helper

if t.TYPE_CHECKING:
    import codeop  # noqa: F401

_stream: ContextVar["HTMLStringO"] = ContextVar("werkzeug.debug.console.stream")
_ipy: ContextVar = ContextVar("werkzeug.debug.console.ipy")

class HTMLStringO:
    """A StringO version that HTML escapes on write."""

def __init__(self) -> None:
        self._buffer: t.List[str] = []

def isatty(self) -> bool:
        return False

def close(self) -> None:
        pass

def flush(self) -> None:
        pass

def seek(self, n: int, mode: int = 0) -> None:
        pass

def readline(self) -> str:
        if len(self._buffer) == 0:
            return ""
        ret = self._buffer[0]
        del self._buffer[0]
        return ret

def reset(self) -> str:
        val = "".join(self._buffer)
        del self._buffer[:]
        return val

def _write(self, x: str) -> None:
        if isinstance(x, bytes):
            x = x.decode("utf-8", "replace")
        self._buffer.append(x)

def write(self, x: str) -> None:
        self._write(escape(x))

def writelines(self, x: t.Iterable[str]) -> None:
        self._write(escape("".join(x)))

class ThreadedStream:
    """Thread-local wrapper for sys.stdout for the interactive console."""

@staticmethod
    def push() -> None:
        if not isinstance(sys.stdout, ThreadedStream):
            sys.stdout = t.cast(t.TextIO, ThreadedStream())

_stream.set(HTMLStringO())

@staticmethod
    def fetch() -> str:
        try:
            stream = _stream.get()
        except LookupError:
            return ""

return stream.reset()

@staticmethod
    def displayhook(obj: object) -> None:
        try:
            stream = _stream.get()
        except LookupError:
            return _displayhook(obj)  # type: ignore

# stream._write bypasses escaping as debug_repr is
        # already generating HTML for us.
        if obj is not None:
            _ipy.get().locals["_"] = obj
            stream._write(debug_repr(obj))

def __setattr__(self, name: str, value: t.Any) -> None:
        raise AttributeError(f"read only attribute {name}")

def __dir__(self) -> t.List[str]:
        return dir(sys.__stdout__)

def __getattribute__(self, name: str) -> t.Any:
        try:
            stream = _stream.get()
        except LookupError:
            stream = sys.__stdout__  # type: ignore[assignment]

return getattr(stream, name)

def __repr__(self) -> str:
        return repr(sys.__stdout__)

# add the threaded stream as display hook
_displayhook = sys.displayhook
sys.displayhook = ThreadedStream.displayhook

class _ConsoleLoader:
    def __init__(self) -> None:
        self._storage: t.Dict[int, str] = {}

def register(self, code: CodeType, source: str) -> None:
        self._storage[id(code)] = source
        # register code objects of wrapped functions too.
        for var in code.co_consts:
            if isinstance(var, CodeType):
                self._storage[id(var)] = source

def get_source_by_code(self, code: CodeType) -> t.Optional[str]:
        try:
            return self._storage[id(code)]
        except KeyError:
            return None

class _InteractiveConsole(code.InteractiveInterpreter):
    locals: t.Dict[str, t.Any]

def __init__(self, globals: t.Dict[str, t.Any], locals: t.Dict[str, t.Any]) -> None:
        self.loader = _ConsoleLoader()
        locals = {
            **globals,
            **locals,
            "dump": dump,
            "help": helper,
            "__loader__": self.loader,
        }
        super().__init__(locals)
        original_compile = self.compile

def compile(source: str, filename: str, symbol: str) -> t.Optional[CodeType]:
            code = original_compile(source, filename, symbol)

if code is not None:
                self.loader.register(code, source)

return code

self.compile = compile  # type: ignore[assignment]
        self.more = False
        self.buffer: t.List[str] = []

def runsource(self, source: str, **kwargs: t.Any) -> str:  # type: ignore
        source = f"{source.rstrip()}\n"
        ThreadedStream.push()
        prompt = "... " if self.more else ">>> "
        try:
            source_to_eval = "".join(self.buffer + [source])
            if super().runsource(source_to_eval, "<debugger>", "single"):
                self.more = True
                self.buffer.append(source)
            else:
                self.more = False
                del self.buffer[:]
        finally:
            output = ThreadedStream.fetch()
        return f"{prompt}{escape(source)}{output}"

def runcode(self, code: CodeType) -> None:
        try:
            exec(code, self.locals)
        except Exception:
            self.showtraceback()

def showtraceback(self) -> None:
        from .tbtools import DebugTraceback

exc = t.cast(BaseException, sys.exc_info()[1])
        te = DebugTraceback(exc, skip=1)
        sys.stdout._write(te.render_traceback_html())  # type: ignore

def showsyntaxerror(self, filename: t.Optional[str] = None) -> None:
        from .tbtools import DebugTraceback

exc = t.cast(BaseException, sys.exc_info()[1])
        te = DebugTraceback(exc, skip=4)
        sys.stdout._write(te.render_traceback_html())  # type: ignore

def write(self, data: str) -> None:
        sys.stdout.write(data)

class Console:
    """An interactive console."""

def __init__(
        self,
        globals: t.Optional[t.Dict[str, t.Any]] = None,
        locals: t.Optional[t.Dict[str, t.Any]] = None,
    ) -> None:
        if locals is None:
            locals = {}
        if globals is None:
            globals = {}
        self._ipy = _InteractiveConsole(globals, locals)

def eval(self, code: str) -> str:
        _ipy.set(self._ipy)
        old_sys_stdout = sys.stdout
        try:
            return self._ipy.runsource(code)
        finally:
            sys.stdout = old_sys_stdout

Veranstaltungen | Fairnetzt-Hochrhein

Keine Veranstaltung gefunden!