diff options
author | monster <monster@ydb.tech> | 2022-07-07 14:41:37 +0300 |
---|---|---|
committer | monster <monster@ydb.tech> | 2022-07-07 14:41:37 +0300 |
commit | 06e5c21a835c0e923506c4ff27929f34e00761c2 (patch) | |
tree | 75efcbc6854ef9bd476eb8bf00cc5c900da436a2 /contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py | |
parent | 03f024c4412e3aa613bb543cf1660176320ba8f4 (diff) | |
download | ydb-06e5c21a835c0e923506c4ff27929f34e00761c2.tar.gz |
fix ya.make
Diffstat (limited to 'contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py')
-rw-r--r-- | contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py | 301 |
1 files changed, 0 insertions, 301 deletions
diff --git a/contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py b/contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py deleted file mode 100644 index 987d7175de..0000000000 --- a/contrib/python/prompt-toolkit/py3/prompt_toolkit/history.py +++ /dev/null @@ -1,301 +0,0 @@ -""" -Implementations for the history of a `Buffer`. - -NOTE: There is no `DynamicHistory`: - This doesn't work well, because the `Buffer` needs to be able to attach - an event handler to the event when a history entry is loaded. This - loading can be done asynchronously and making the history swappable would - probably break this. -""" -import datetime -import os -import threading -from abc import ABCMeta, abstractmethod -from typing import AsyncGenerator, Iterable, List, Optional, Sequence, Tuple - -from prompt_toolkit.eventloop import get_event_loop - -__all__ = [ - "History", - "ThreadedHistory", - "DummyHistory", - "FileHistory", - "InMemoryHistory", -] - - -class History(metaclass=ABCMeta): - """ - Base ``History`` class. - - This also includes abstract methods for loading/storing history. - """ - - def __init__(self) -> None: - # In memory storage for strings. - self._loaded = False - - # History that's loaded already, in reverse order. Latest, most recent - # item first. - self._loaded_strings: List[str] = [] - - # - # Methods expected by `Buffer`. - # - - async def load(self) -> AsyncGenerator[str, None]: - """ - Load the history and yield all the entries in reverse order (latest, - most recent history entry first). - - This method can be called multiple times from the `Buffer` to - repopulate the history when prompting for a new input. So we are - responsible here for both caching, and making sure that strings that - were were appended to the history will be incorporated next time this - method is called. - """ - if not self._loaded: - self._loaded_strings = list(self.load_history_strings()) - self._loaded = True - - for item in self._loaded_strings: - yield item - - def get_strings(self) -> List[str]: - """ - Get the strings from the history that are loaded so far. - (In order. Oldest item first.) - """ - return self._loaded_strings[::-1] - - def append_string(self, string: str) -> None: - "Add string to the history." - self._loaded_strings.insert(0, string) - self.store_string(string) - - # - # Implementation for specific backends. - # - - @abstractmethod - def load_history_strings(self) -> Iterable[str]: - """ - This should be a generator that yields `str` instances. - - It should yield the most recent items first, because they are the most - important. (The history can already be used, even when it's only - partially loaded.) - """ - while False: - yield - - @abstractmethod - def store_string(self, string: str) -> None: - """ - Store the string in persistent storage. - """ - - -class ThreadedHistory(History): - """ - Wrapper around `History` implementations that run the `load()` generator in - a thread. - - Use this to increase the start-up time of prompt_toolkit applications. - History entries are available as soon as they are loaded. We don't have to - wait for everything to be loaded. - """ - - def __init__(self, history: History) -> None: - super().__init__() - - self.history = history - - self._load_thread: Optional[threading.Thread] = None - - # Lock for accessing/manipulating `_loaded_strings` and `_loaded` - # together in a consistent state. - self._lock = threading.Lock() - - # Events created by each `load()` call. Used to wait for new history - # entries from the loader thread. - self._string_load_events: List[threading.Event] = [] - - async def load(self) -> AsyncGenerator[str, None]: - """ - Like `History.load(), but call `self.load_history_strings()` in a - background thread. - """ - # Start the load thread, if this is called for the first time. - if not self._load_thread: - self._load_thread = threading.Thread( - target=self._in_load_thread, - daemon=True, - ) - self._load_thread.start() - - # Consume the `_loaded_strings` list, using asyncio. - loop = get_event_loop() - - # Create threading Event so that we can wait for new items. - event = threading.Event() - event.set() - self._string_load_events.append(event) - - items_yielded = 0 - - try: - while True: - # Wait for new items to be available. - # (Use a timeout, because the executor thread is not a daemon - # thread. The "slow-history.py" example would otherwise hang if - # Control-C is pressed before the history is fully loaded, - # because there's still this non-daemon executor thread waiting - # for this event.) - got_timeout = await loop.run_in_executor( - None, lambda: event.wait(timeout=0.5) - ) - if not got_timeout: - continue - - # Read new items (in lock). - def in_executor() -> Tuple[List[str], bool]: - with self._lock: - new_items = self._loaded_strings[items_yielded:] - done = self._loaded - event.clear() - return new_items, done - - new_items, done = await loop.run_in_executor(None, in_executor) - - items_yielded += len(new_items) - - for item in new_items: - yield item - - if done: - break - finally: - self._string_load_events.remove(event) - - def _in_load_thread(self) -> None: - try: - # Start with an empty list. In case `append_string()` was called - # before `load()` happened. Then `.store_string()` will have - # written these entries back to disk and we will reload it. - self._loaded_strings = [] - - for item in self.history.load_history_strings(): - with self._lock: - self._loaded_strings.append(item) - - for event in self._string_load_events: - event.set() - finally: - with self._lock: - self._loaded = True - for event in self._string_load_events: - event.set() - - def append_string(self, string: str) -> None: - with self._lock: - self._loaded_strings.insert(0, string) - self.store_string(string) - - # All of the following are proxied to `self.history`. - - def load_history_strings(self) -> Iterable[str]: - return self.history.load_history_strings() - - def store_string(self, string: str) -> None: - self.history.store_string(string) - - def __repr__(self) -> str: - return f"ThreadedHistory({self.history!r})" - - -class InMemoryHistory(History): - """ - :class:`.History` class that keeps a list of all strings in memory. - - In order to prepopulate the history, it's possible to call either - `append_string` for all items or pass a list of strings to `__init__` here. - """ - - def __init__(self, history_strings: Optional[Sequence[str]] = None) -> None: - super().__init__() - # Emulating disk storage. - if history_strings is None: - self._storage = [] - else: - self._storage = list(history_strings) - - def load_history_strings(self) -> Iterable[str]: - yield from self._storage[::-1] - - def store_string(self, string: str) -> None: - self._storage.append(string) - - -class DummyHistory(History): - """ - :class:`.History` object that doesn't remember anything. - """ - - def load_history_strings(self) -> Iterable[str]: - return [] - - def store_string(self, string: str) -> None: - pass - - def append_string(self, string: str) -> None: - # Don't remember this. - pass - - -class FileHistory(History): - """ - :class:`.History` class that stores all strings in a file. - """ - - def __init__(self, filename: str) -> None: - self.filename = filename - super().__init__() - - def load_history_strings(self) -> Iterable[str]: - strings: List[str] = [] - lines: List[str] = [] - - def add() -> None: - if lines: - # Join and drop trailing newline. - string = "".join(lines)[:-1] - - strings.append(string) - - if os.path.exists(self.filename): - with open(self.filename, "rb") as f: - for line_bytes in f: - line = line_bytes.decode("utf-8", errors="replace") - - if line.startswith("+"): - lines.append(line[1:]) - else: - add() - lines = [] - - add() - - # Reverse the order, because newest items have to go first. - return reversed(strings) - - def store_string(self, string: str) -> None: - # Save to file. - with open(self.filename, "ab") as f: - - def write(t: str) -> None: - f.write(t.encode("utf-8")) - - write("\n# %s\n" % datetime.datetime.now()) - for line in string.split("\n"): - write("+%s\n" % line) |