summaryrefslogtreecommitdiffstats
path: root/src/prompt_toolkit/application/current.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 17:35:20 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 17:35:20 +0000
commite106bf94eff07d9a59771d9ccc4406421e18ab64 (patch)
treeedb6545500e39df9c67aa918a6125bffc8ec1aee /src/prompt_toolkit/application/current.py
parentInitial commit. (diff)
downloadprompt-toolkit-upstream/3.0.36.tar.xz
prompt-toolkit-upstream/3.0.36.zip
Adding upstream version 3.0.36.upstream/3.0.36upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r--src/prompt_toolkit/application/current.py197
1 files changed, 197 insertions, 0 deletions
diff --git a/src/prompt_toolkit/application/current.py b/src/prompt_toolkit/application/current.py
new file mode 100644
index 0000000..ae69bfd
--- /dev/null
+++ b/src/prompt_toolkit/application/current.py
@@ -0,0 +1,197 @@
+import sys
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any, Generator, Optional
+
+try:
+ from contextvars import ContextVar
+except ImportError:
+ from prompt_toolkit.eventloop.dummy_contextvars import ContextVar # type: ignore
+
+if TYPE_CHECKING:
+ from prompt_toolkit.input.base import Input
+ from prompt_toolkit.output.base import Output
+
+ from .application import Application
+
+__all__ = [
+ "AppSession",
+ "get_app_session",
+ "get_app",
+ "get_app_or_none",
+ "set_app",
+ "create_app_session",
+ "create_app_session_from_tty",
+]
+
+
+class AppSession:
+ """
+ An AppSession is an interactive session, usually connected to one terminal.
+ Within one such session, interaction with many applications can happen, one
+ after the other.
+
+ The input/output device is not supposed to change during one session.
+
+ Warning: Always use the `create_app_session` function to create an
+ instance, so that it gets activated correctly.
+
+ :param input: Use this as a default input for all applications
+ running in this session, unless an input is passed to the `Application`
+ explicitely.
+ :param output: Use this as a default output.
+ """
+
+ def __init__(
+ self, input: Optional["Input"] = None, output: Optional["Output"] = None
+ ) -> None:
+
+ self._input = input
+ self._output = output
+
+ # The application will be set dynamically by the `set_app` context
+ # manager. This is called in the application itself.
+ self.app: Optional["Application[Any]"] = None
+
+ def __repr__(self) -> str:
+ return f"AppSession(app={self.app!r})"
+
+ @property
+ def input(self) -> "Input":
+ if self._input is None:
+ from prompt_toolkit.input.defaults import create_input
+
+ self._input = create_input()
+ return self._input
+
+ @property
+ def output(self) -> "Output":
+ if self._output is None:
+ from prompt_toolkit.output.defaults import create_output
+
+ self._output = create_output()
+ return self._output
+
+
+_current_app_session: ContextVar["AppSession"] = ContextVar(
+ "_current_app_session", default=AppSession()
+)
+
+
+def get_app_session() -> AppSession:
+ return _current_app_session.get()
+
+
+def get_app() -> "Application[Any]":
+ """
+ Get the current active (running) Application.
+ An :class:`.Application` is active during the
+ :meth:`.Application.run_async` call.
+
+ We assume that there can only be one :class:`.Application` active at the
+ same time. There is only one terminal window, with only one stdin and
+ stdout. This makes the code significantly easier than passing around the
+ :class:`.Application` everywhere.
+
+ If no :class:`.Application` is running, then return by default a
+ :class:`.DummyApplication`. For practical reasons, we prefer to not raise
+ an exception. This way, we don't have to check all over the place whether
+ an actual `Application` was returned.
+
+ (For applications like pymux where we can have more than one `Application`,
+ we'll use a work-around to handle that.)
+ """
+ session = _current_app_session.get()
+ if session.app is not None:
+ return session.app
+
+ from .dummy import DummyApplication
+
+ return DummyApplication()
+
+
+def get_app_or_none() -> Optional["Application[Any]"]:
+ """
+ Get the current active (running) Application, or return `None` if no
+ application is running.
+ """
+ session = _current_app_session.get()
+ return session.app
+
+
+@contextmanager
+def set_app(app: "Application[Any]") -> Generator[None, None, None]:
+ """
+ Context manager that sets the given :class:`.Application` active in an
+ `AppSession`.
+
+ This should only be called by the `Application` itself.
+ The application will automatically be active while its running. If you want
+ the application to be active in other threads/coroutines, where that's not
+ the case, use `contextvars.copy_context()`, or use `Application.context` to
+ run it in the appropriate context.
+ """
+ session = _current_app_session.get()
+
+ previous_app = session.app
+ session.app = app
+ try:
+ yield
+ finally:
+ session.app = previous_app
+
+
+@contextmanager
+def create_app_session(
+ input: Optional["Input"] = None, output: Optional["Output"] = None
+) -> Generator[AppSession, None, None]:
+ """
+ Create a separate AppSession.
+
+ This is useful if there can be multiple individual `AppSession`s going on.
+ Like in the case of an Telnet/SSH server. This functionality uses
+ contextvars and requires at least Python 3.7.
+ """
+ if sys.version_info <= (3, 6):
+ raise RuntimeError("Application sessions require Python 3.7.")
+
+ # If no input/output is specified, fall back to the current input/output,
+ # whatever that is.
+ if input is None:
+ input = get_app_session().input
+ if output is None:
+ output = get_app_session().output
+
+ # Create new `AppSession` and activate.
+ session = AppSession(input=input, output=output)
+
+ token = _current_app_session.set(session)
+ try:
+ yield session
+ finally:
+ _current_app_session.reset(token)
+
+
+@contextmanager
+def create_app_session_from_tty() -> Generator[AppSession, None, None]:
+ """
+ Create `AppSession` that always prefers the TTY input/output.
+
+ Even if `sys.stdin` and `sys.stdout` are connected to input/output pipes,
+ this will still use the terminal for interaction (because `sys.stderr` is
+ still connected to the terminal).
+
+ Usage::
+
+ from prompt_toolkit.shortcuts import prompt
+
+ with create_app_session_from_tty():
+ prompt('>')
+ """
+ from prompt_toolkit.input.defaults import create_input
+ from prompt_toolkit.output.defaults import create_output
+
+ input = create_input(always_prefer_tty=True)
+ output = create_output(always_prefer_tty=True)
+
+ with create_app_session(input=input, output=output) as app_session:
+ yield app_session