1from __future__ import annotations
2
3from collections.abc import Generator
4from contextlib import contextmanager
5from contextvars import ContextVar
6from typing import TYPE_CHECKING, Any
7
8if TYPE_CHECKING:
9 from prompt_toolkit.input.base import Input
10 from prompt_toolkit.output.base import Output
11
12 from .application import Application
13
14__all__ = [
15 "AppSession",
16 "get_app_session",
17 "get_app",
18 "get_app_or_none",
19 "set_app",
20 "create_app_session",
21 "create_app_session_from_tty",
22]
23
24
25class AppSession:
26 """
27 An AppSession is an interactive session, usually connected to one terminal.
28 Within one such session, interaction with many applications can happen, one
29 after the other.
30
31 The input/output device is not supposed to change during one session.
32
33 Warning: Always use the `create_app_session` function to create an
34 instance, so that it gets activated correctly.
35
36 :param input: Use this as a default input for all applications
37 running in this session, unless an input is passed to the `Application`
38 explicitly.
39 :param output: Use this as a default output.
40 """
41
42 def __init__(
43 self, input: Input | None = None, output: Output | None = None
44 ) -> None:
45 self._input = input
46 self._output = output
47
48 # The application will be set dynamically by the `set_app` context
49 # manager. This is called in the application itself.
50 self.app: Application[Any] | None = None
51
52 def __repr__(self) -> str:
53 return f"AppSession(app={self.app!r})"
54
55 @property
56 def input(self) -> Input:
57 if self._input is None:
58 from prompt_toolkit.input.defaults import create_input
59
60 self._input = create_input()
61 return self._input
62
63 @property
64 def output(self) -> Output:
65 if self._output is None:
66 from prompt_toolkit.output.defaults import create_output
67
68 self._output = create_output()
69 return self._output
70
71
72_current_app_session: ContextVar[AppSession] = ContextVar(
73 "_current_app_session", default=AppSession()
74)
75
76
77def get_app_session() -> AppSession:
78 return _current_app_session.get()
79
80
81def get_app() -> Application[Any]:
82 """
83 Get the current active (running) Application.
84 An :class:`.Application` is active during the
85 :meth:`.Application.run_async` call.
86
87 We assume that there can only be one :class:`.Application` active at the
88 same time. There is only one terminal window, with only one stdin and
89 stdout. This makes the code significantly easier than passing around the
90 :class:`.Application` everywhere.
91
92 If no :class:`.Application` is running, then return by default a
93 :class:`.DummyApplication`. For practical reasons, we prefer to not raise
94 an exception. This way, we don't have to check all over the place whether
95 an actual `Application` was returned.
96
97 (For applications like pymux where we can have more than one `Application`,
98 we'll use a work-around to handle that.)
99 """
100 session = _current_app_session.get()
101 if session.app is not None:
102 return session.app
103
104 from .dummy import DummyApplication
105
106 return DummyApplication()
107
108
109def get_app_or_none() -> Application[Any] | None:
110 """
111 Get the current active (running) Application, or return `None` if no
112 application is running.
113 """
114 session = _current_app_session.get()
115 return session.app
116
117
118@contextmanager
119def set_app(app: Application[Any]) -> Generator[None, None, None]:
120 """
121 Context manager that sets the given :class:`.Application` active in an
122 `AppSession`.
123
124 This should only be called by the `Application` itself.
125 The application will automatically be active while its running. If you want
126 the application to be active in other threads/coroutines, where that's not
127 the case, use `contextvars.copy_context()`, or use `Application.context` to
128 run it in the appropriate context.
129 """
130 session = _current_app_session.get()
131
132 previous_app = session.app
133 session.app = app
134 try:
135 yield
136 finally:
137 session.app = previous_app
138
139
140@contextmanager
141def create_app_session(
142 input: Input | None = None, output: Output | None = None
143) -> Generator[AppSession, None, None]:
144 """
145 Create a separate AppSession.
146
147 This is useful if there can be multiple individual ``AppSession``'s going
148 on. Like in the case of a Telnet/SSH server.
149 """
150 # If no input/output is specified, fall back to the current input/output,
151 # if there was one that was set/created for the current session.
152 # (Note that we check `_input`/`_output` and not `input`/`output`. This is
153 # because we don't want to accidentally create a new input/output objects
154 # here and store it in the "parent" `AppSession`. Especially, when
155 # combining pytest's `capsys` fixture and `create_app_session`, sys.stdin
156 # and sys.stderr are patched for every test, so we don't want to leak
157 # those outputs object across `AppSession`s.)
158 if input is None:
159 input = get_app_session()._input
160 if output is None:
161 output = get_app_session()._output
162
163 # Create new `AppSession` and activate.
164 session = AppSession(input=input, output=output)
165
166 token = _current_app_session.set(session)
167 try:
168 yield session
169 finally:
170 _current_app_session.reset(token)
171
172
173@contextmanager
174def create_app_session_from_tty() -> Generator[AppSession, None, None]:
175 """
176 Create `AppSession` that always prefers the TTY input/output.
177
178 Even if `sys.stdin` and `sys.stdout` are connected to input/output pipes,
179 this will still use the terminal for interaction (because `sys.stderr` is
180 still connected to the terminal).
181
182 Usage::
183
184 from prompt_toolkit.shortcuts import prompt
185
186 with create_app_session_from_tty():
187 prompt('>')
188 """
189 from prompt_toolkit.input.defaults import create_input
190 from prompt_toolkit.output.defaults import create_output
191
192 input = create_input(always_prefer_tty=True)
193 output = create_output(always_prefer_tty=True)
194
195 with create_app_session(input=input, output=output) as app_session:
196 yield app_session