Source code for xonsh.ptk_shell.formatter

"""PTK specific PromptFormatter class."""

import functools
import typing as tp

from xonsh.prompt.base import DEFAULT_PROMPT, PromptFormatter
from xonsh.ptk_shell.updator import AsyncPrompt, PromptUpdator


[docs]class PTKPromptFormatter(PromptFormatter): """A subclass of PromptFormatter to support rendering prompt sections with/without threads.""" def __init__(self, shell): super().__init__() self.shell = shell def __call__( self, template=DEFAULT_PROMPT, fields=None, threaded=False, prompt_name: str = None, **_, ) -> str: """Formats a xonsh prompt template string.""" kwargs = {} if threaded: # init only for async prompts if not hasattr(self, "updator"): # updates an async prompt. self.updator = PromptUpdator(self.shell) # set these attributes per call. one can enable/disable async-prompt inside a session. kwargs["async_prompt"] = self.updator.add(prompt_name) # in case of failure it returns a fail-over template. otherwise it returns list of tokens return super().__call__(template, fields, **kwargs) def _format_prompt( self, template=DEFAULT_PROMPT, async_prompt: tp.Optional[AsyncPrompt] = None, **kwargs, ): toks = super()._format_prompt( template=template, async_prompt=async_prompt, **kwargs ) if async_prompt is not None: # late binding of values async_prompt.tokens = toks return toks def _get_field_value( self, field, async_prompt=None, idx=None, spec=None, conv=None, **_ ): func = functools.partial(super()._get_field_value, field) if async_prompt is not None and self.fields.needs_calling(field): # create a thread and return an intermediate result return async_prompt.submit_section(func, field, idx, spec, conv) return func()
[docs] def start_update(self): """Start listening on the prompt section futures.""" self.updator.start()