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