深入Python Prompt Toolkit:从基础提示到高级功能

【免费下载链接】python-prompt-toolkit Library for building powerful interactive command line applications in Python 【免费下载链接】python-prompt-toolkit 项目地址: https://gitcode.com/gh_mirrors/py/python-prompt-toolkit

本文深入探讨Python Prompt Toolkit库的核心功能,从基础的PromptSession对象配置到高级的语法高亮、自动补全、输入验证和历史记录管理。PromptSession作为核心组件,提供了强大的会话状态管理能力,支持丰富的配置选项和自定义功能。文章详细解析了Pygments集成实现语法高亮、多种自动补全器的实现方式,以及输入验证和历史记录的高级应用技巧,为开发者构建现代化命令行应用提供全面指导。

PromptSession对象详解与配置选项

PromptSession是Python Prompt Toolkit库的核心组件,它提供了一个强大的交互式命令行会话管理机制。与简单的prompt()函数不同,PromptSession能够维护会话状态,包括历史记录、配置选项和用户偏好,为构建复杂的命令行应用提供了坚实的基础。

PromptSession的核心优势

PromptSession相比一次性提示函数的主要优势在于其状态保持能力:

mermaid

构造函数参数详解

PromptSession的构造函数提供了丰富的配置选项,可以分为以下几个主要类别:

基本显示配置
参数 类型 默认值 描述
message AnyFormattedText "" 提示消息,支持格式化文本
multiline FilterOrBool False 是否启用多行输入模式
wrap_lines FilterOrBool True 是否自动换行
is_password FilterOrBool False 是否为密码输入
编辑模式配置
from prompt_toolkit.enums import EditingMode

# 配置VI编辑模式
session = PromptSession(editing_mode=EditingMode.VI)

# 或者使用简化的vi_mode参数
session = PromptSession(vi_mode=True)
自动完成与建议

mermaid

样式与外观配置
from prompt_toolkit.styles import Style
from prompt_toolkit.lexers import PygmentsLexer
from pygments.lexers.python import PythonLexer

# 创建自定义样式
custom_style = Style.from_dict({
    'prompt': 'bg:blue #ffffff',
    'input': '#ff0066',
    'completion-menu.completion': 'bg:#008888 #ffffff',
})

session = PromptSession(
    style=custom_style,
    lexer=PygmentsLexer(PythonLexer),
    color_depth="DEPTH_24_BIT"  # 真彩色支持
)
历史记录管理

PromptSession支持多种历史记录后端:

from prompt_toolkit.history import InMemoryHistory, FileHistory

# 内存历史记录
memory_history = InMemoryHistory()
memory_history.append_string("command1")
memory_history.append_string("command2")

# 文件历史记录
file_history = FileHistory("~/.myapp_history")

session = PromptSession(history=file_history)

核心方法解析

prompt() 方法

prompt()方法是PromptSession的核心功能,它支持覆盖会话级别的配置:

# 基础用法
result = session.prompt()

# 覆盖会话配置
result = session.prompt(
    message="临时提示: ",
    default="默认值",
    is_password=True  # 本次提示为密码输入
)
prompt_async() 方法

对于异步应用,PromptSession提供了异步版本的提示方法:

async def async_example():
    session = PromptSession()
    result = await session.prompt_async("异步提示: ")
    return result

高级配置选项

输入验证器
from prompt_toolkit.validation import Validator, ValidationError
from prompt_toolkit.document import Document

class NumberValidator(Validator):
    def validate(self, document: Document):
        text = document.text
        if text and not text.isdigit():
            raise ValidationError(
                message="请输入数字",
                cursor_position=len(text)
            )

session = PromptSession(validator=NumberValidator())
自定义键绑定
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.keys import Keys

kb = KeyBindings()

@kb.add(Keys.ControlT)
def _(event):
    "插入当前时间"
    event.app.current_buffer.insert_text("2024-01-01")

session = PromptSession(key_bindings=kb)
多语言支持与处理器
from prompt_toolkit.layout.processors import (
    HighlightSelectionProcessor,
    AppendAutoSuggestion,
    PasswordProcessor
)

session = PromptSession(
    input_processors=[
        HighlightSelectionProcessor(),  # 高亮选中文本
        AppendAutoSuggestion(),         # 自动建议
        PasswordProcessor() if is_password else None  # 条件处理器
    ]
)

实际应用示例

数据库CLI工具
import sqlite3
from prompt_toolkit import PromptSession
from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.lexers import PygmentsLexer
from pygments.lexers.sql import SqlLexer

class DatabaseCLI:
    def __init__(self, db_path):
        self.connection = sqlite3.connect(db_path)
        self.completer = WordCompleter([
            'SELECT', 'FROM', 'WHERE', 'INSERT', 'UPDATE', 'DELETE'
        ], ignore_case=True)
        
        self.session = PromptSession(
            lexer=PygmentsLexer(SqlLexer),
            completer=self.completer,
            multiline=True,  # SQL语句通常需要多行
            prompt_continuation=lambda width, line_number, wrap_count: 
                '... ' if line_number > 0 else 'SQL> '
        )
    
    def run(self):
        while True:
            try:
                query = self.session.prompt()
                if query.strip().lower() == 'quit':
                    break
                
                cursor = self.connection.execute(query)
                for row in cursor:
                    print(row)
                    
            except Exception as e:
                print(f"错误: {e}")
配置管理会话
from prompt_toolkit.styles import Style
from prompt_toolkit.formatted_text import HTML

class ConfigSession:
    def __init__(self):
        self.style = Style.from_dict({
            'header': 'bg:blue #ffffff bold',
            'prompt': '#00ff00',
            'input': '#ff69b4',
        })
        
        self.session = PromptSession(
            style=self.style,
            bottom_toolbar=self._get_toolbar,
            enable_system_prompt=True
        )
    
    def _get_toolbar(self):
        return HTML(
            '<b>[F1]</b> Help <b>[F2]</b> Settings <b>[Ctrl+D]</b> Exit'
        )
    
    def get_input(self, prompt_text):
        # 使用HTML格式化提示
        formatted_prompt = HTML(f'<prompt>{prompt_text}</prompt>')
        return self.session.prompt(formatted_prompt)

性能优化技巧

延迟加载配置

对于大型应用,可以使用动态配置来优化性能:

session = PromptSession(
    completer=DynamicCompleter(lambda: self._get_completer()),
    style=DynamicStyle(lambda: self._get_current_style()),
    auto_suggest=DynamicAutoSuggest(lambda: self._get_suggestions())
)
线程安全处理
from threading import Lock

class ThreadSafeSession:
    def __init__(self):
        self.session = PromptSession()
        self.lock = Lock()
    
    def prompt(self, *args, **kwargs):
        with self.lock:
            return self.session.prompt(*args, **kwargs)

错误处理与调试

异常处理策略
try:
    result = session.prompt("输入: ")
except KeyboardInterrupt:
    print("操作被用户取消")
except EOFError:
    print("退出应用")
except Exception as e:
    print(f"未知错误: {e}")
调试模式配置
debug_session = PromptSession(
    refresh_interval=0.1,  # 更快的刷新频率用于调试
    mouse_support=True,    # 启用鼠标支持便于测试
    tempfile_suffix=".debug"  # 调试临时文件后缀
)

PromptSession的丰富配置选项和灵活的设计使其成为构建现代化命令行应用的理想选择。通过合理利用其各种功能,可以创建出既美观又功能强大的交互式命令行工具。

语法高亮与Pygments集成使用

在构建交互式命令行应用时,语法高亮是提升用户体验的重要功能。Python Prompt Toolkit通过深度集成Pygments库,为开发者提供了强大的语法高亮能力。Pygments是Python生态中最著名的语法高亮库,支持超过500种编程语言和标记语言的语法分析。

PygmentsLexer核心架构

PygmentsLexer是prompt_toolkit与Pygments之间的桥梁,它实现了高效的语法分析和样式映射机制。其核心架构基于以下几个关键组件:

mermaid

基本使用方法

使用PygmentsLexer非常简单,只需几行代码即可为输入添加语法高亮:

from pygments.lexers.html import HtmlLexer
from prompt_toolkit import prompt
from prompt_toolkit.lexers import PygmentsLexer

# 基本语法高亮
text = prompt("Enter HTML: ", lexer=PygmentsLexer(HtmlLexer))
print(f"You entered: {text}")

自动文件类型检测

PygmentsLexer提供了智能的文件类型检测功能,可以根据文件扩展名自动选择合适的语法分析器:

from prompt_toolkit.lexers import PygmentsLexer
from prompt_toolkit.widgets import TextArea

# 动态语法高亮基于文件扩展名
text_area = TextArea(
    lexer=DynamicLexer(
        lambda: PygmentsLexer.from_filename(
            file_path or ".txt", 
            sync_from_start=False
        )
    )
)

语法同步优化

为了提高大文件的处理性能,PygmentsLexer实现了智能的语法同步机制:

from prompt_toolkit.lexers import PygmentsLexer, RegexSync
from pygments.lexers.python import PythonLexer

# 自定义同步策略
custom_sync = RegexSync(r"^\s*(class|def|async def)\s+")
lexer = PygmentsLexer(
    PythonLexer, 
    sync_from_start=False,
    syntax_sync=custom_sync
)

PygmentsLexer内置了多种语言的同步模式:

语言 同步模式正则表达式 说明
Python ^\s*(class\|def)\s+ 类和方法定义
HTML <[/a-zA-Z] 标签开始
JavaScript \bfunction\b 函数定义
默认 ^ 每行开始

样式配置与自定义

语法高亮的样式可以通过Pygments样式系统进行配置:

from prompt_toolkit.styles import Style
from prompt_toolkit.styles.pygments import style_from_pygments_cls
from pygments.styles import get_style_by_name

# 使用内置Pygments样式
style = style_from_pygments_cls(get_style_by_name('monokai'))

# 自定义样式映射
custom_style = Style.from_dict({
    "pygments.keyword": "bold #ff0000",
    "pygments.comment": "italic #888888",
    "pygments.string": "#00ff00",
    "pygments.number": "#ff00ff",
})

高级用法示例

以下是一个完整的高级示例,展示了如何结合多种功能:

from pygments.lexers.python import PythonLexer
from prompt_toolkit import PromptSession
from prompt_toolkit.lexers import PygmentsLexer
from prompt_toolkit.styles import Style
from prompt_toolkit.completion import WordCompleter

# 创建自定义样式
style = Style.from_dict({
    "pygments.keyword": "bold ansired",
    "pygments.comment": "italic #888888",
    "pygments.string": "ansigreen",
    "pygments.number": "ansimagenta",
    "pygments.name.function": "ansiblue",
})

# Python关键字自动补全
python_completer = WordCompleter([
    'def', 'class', 'import', 'from', 'as', 'if', 'else',
    'elif', 'for', 'while', 'return', 'yield', 'async', 'await'
])

# 创建支持语法高亮的会话
session = PromptSession(
    lexer=PygmentsLexer(PythonLexer),
    style=style,
    completer=python_completer,
    include_default_pygments_style=True
)

while True:
    try:
        code = session.prompt("Python >>> ")
        print(f"Executing: {code}")
    except KeyboardInterrupt:
        continue
    except EOFError:
        break

性能优化策略

对于大型文件或实时编辑场景,性能优化至关重要:

  1. 语法同步: 避免从文件开头重新分析,使用智能同步点
  2. 缓存机制: 对已分析的文本行进行缓存
  3. 延迟分析: 只在需要时进行分析,减少不必要的计算
# 性能优化配置
optimized_lexer = PygmentsLexer(
    PythonLexer,
    sync_from_start=False,  # 禁用从头开始分析
    syntax_sync=RegexSync.from_pygments_lexer_cls(PythonLexer)
)

错误处理与回退机制

PygmentsLexer提供了完善的错误处理机制:

from prompt_toolkit.lexers import PygmentsLexer, SimpleLexer

try:
    # 尝试获取特定语言的语法分析器
    lexer = PygmentsLexer.from_filename("example.unknown")
except Exception:
    # 回退到简单分析器
    lexer = SimpleLexer()

实际应用场景

PygmentsLexer在多种场景下都有广泛应用:

  1. 交互式解释器: 如IPython、ptpython等
  2. 代码编辑器: 命令行代码编辑器
  3. 配置编辑器: 编辑JSON、YAML、XML等配置文件
  4. 数据查看器: 语法高亮显示数据结构
# 多语言支持示例
from pygments.lexers import (
    PythonLexer, JavascriptLexer, HtmlLexer, 
    JsonLexer, YamlLexer, SqlLexer
)

lexers = {
    '.py': PythonLexer,
    '.js': JavascriptLexer, 
    '.html': HtmlLexer,
    '.json': JsonLexer,
    '.yaml': YamlLexer,
    '.sql': SqlLexer
}

def get_lexer_for_file(filename):
    ext = os.path.splitext(filename)[1].lower()
    return PygmentsLexer(lexers.get(ext, SimpleLexer))

样式继承与覆盖

Pygments样式系统支持灵活的继承和覆盖机制:

mermaid

通过深度集成Pygments,Python Prompt Toolkit为开发者提供了企业级的语法高亮解决方案,无论是简单的命令行提示还是复杂的全屏编辑器,都能获得出色的语法高亮体验。

自动补全机制与自定义实现

Python Prompt Toolkit 提供了强大而灵活的自动补全机制,让开发者能够为命令行应用程序创建智能的输入提示体验。本节将深入探讨其自动补全系统的核心架构、内置补全器类型以及如何实现自定义补全逻辑。

自动补全核心架构

Prompt Toolkit 的自动补全系统基于抽象基类 Completer 构建,所有补全器都必须实现 get_completions 方法。这个架构设计使得补全逻辑可以高度定制化,同时保持接口的一致性。

mermaid

内置补全器类型

Prompt Toolkit 提供了多种内置补全器,覆盖了常见的应用场景:

1. WordCompleter - 单词补全器

WordCompleter 是最基础的补全器,用于基于预定义单词列表的补全:

from prompt_toolkit.completion import WordCompleter

# 简单的单词列表补全
animal_completer = WordCompleter([
    "alligator", "ant", "ape", "bat", "bear", "beaver",
    "bee", "bison", "butterfly", "cat", "chicken"
], ignore_case=True)

# 支持动态获取单词列表
def get_dynamic_words():
    return ["dynamic_word1", "dynamic_word2", "dynamic_word3"]

dynamic_completer = WordCompleter(get_dynamic_words)

配置参数说明:

参数 类型 说明 默认值
words List[str] 或 Callable 补全单词列表或获取函数 必填
ignore_case bool 是否忽略大小写 False
display_dict Mapping[str, AnyFormattedText] 自定义显示格式 None
meta_dict Mapping[str, AnyFormattedText] 元信息显示 None
WORD bool 使用WORD边界 False
sentence bool 句子模式 False
match_middle bool 匹配单词中间 False
2. FuzzyCompleter - 模糊匹配补全器

FuzzyCompleter 提供了强大的模糊匹配功能,允许用户输入部分字符即可匹配相关结果:

from prompt_toolkit.completion import FuzzyCompleter, WordCompleter

# 包装现有补全器实现模糊匹配
fuzzy_animal_completer = FuzzyCompleter(animal_completer)

# 直接使用FuzzyWordCompleter
from prompt_toolkit.completion import FuzzyWordCompleter
fuzzy_completer = FuzzyWordCompleter(["red", "blue", "green", "orange"])

模糊匹配算法基于编辑距离和字符序列匹配,提供智能的排序和过滤功能。

3. PathCompleter - 路径补全器

PathCompleter 专门用于文件和目录路径的自动补全:

from prompt_toolkit.completion import PathCompleter

# 基本路径补全
path_completer = PathCompleter()

# 只补全目录
dir_completer = PathCompleter(only_directories=True)

# 自定义路径过滤器
def python_files_only(path):
    return path.endswith('.py') or os.path.isdir(path)

python_completer = PathCompleter(file_filter=python_files_only)
4. NestedCompleter - 嵌套补全器

NestedCompleter 支持命令层级结构的补全,非常适合CLI工具:

from prompt_toolkit.completion import NestedCompleter, WordCompleter

# 创建嵌套命令结构
nested_completer = NestedCompleter.from_nested_dict({
    'show': {
        'version': None,
        'interfaces': None,
        'configuration': None,
    },
    'configure': {
        'terminal': None,
        'interface': WordCompleter(['ethernet0', 'ethernet1']),
    },
    'exit': None,
    'help': None,
})

自定义补全器实现

创建自定义补全器需要继承 Completer 基类并实现 get_completions 方法:

基础自定义补全器
from prompt_toolkit.completion import Completer, Completion
from prompt_toolkit.document import Document

class CustomCompleter(Completer):
    def __init__(self, options):
        self.options = options
    
    def get_completions(self, document: Document, complete_event):
        word = document.get_word_before_cursor()
        
        for option in self.options:
            if option.startswith(word):
                yield Completion(
                    option,
                    start_position=-len(word),
                    display=option.upper(),  # 自定义显示格式
                    display_meta=f"Custom option: {option}",
                    style="fg:green",
                    selected_style="fg:white bg:green"
                )
高级示例:SQL关键字补全器
class SQLCompleter(Completer):
    def __init__(self):
        self.keywords = [
            'SELECT', 'FROM', 'WHERE', 'INSERT', 'UPDATE', 'DELETE',
            'CREATE', 'ALTER', 'DROP', 'TABLE', 'DATABASE', 'INDEX'
        ]
        self.functions = ['COUNT', 'SUM', 'AVG', 'MAX', 'MIN']
    
    def get_completions(self, document, complete_event):
        text = document.text_before_cursor
        word = document.get_word_before_cursor().upper()
        
        # 根据上下文提供不同的补全建议
        if 'SELECT' in text.upper() and 'FROM' not in text.upper():
            # 在SELECT之后提供函数补全
            for func in self.functions:
                if func.startswith(word):
                    yield Completion(
                        func,
                        start_position=-len(word),
                        display_meta=f"SQL Function: {func}()"
                    )
        else:
            # 提供关键字补全
            for keyword in self.keywords:
                if keyword.startswith(word):
                    yield Completion(
                        keyword,
                        start_position=-len(word),
                        display_meta=f"SQL Keyword: {keyword}"
                    )

补全器组合与装饰

Prompt Toolkit 提供了多种方式来组合和装饰补全器:

1. 合并多个补全器
from prompt_toolkit.completion import merge_completers

# 合并不同类型的补全器
combined_completer = merge_completers([
    keyword_completer,
    function_completer,
    table_name_completer
])
2. 条件补全器
from prompt_toolkit.completion import ConditionalCompleter
from prompt_toolkit.filters import Condition

def is_sql_mode():
    # 根据应用状态决定是否启用补全
    return True

conditional_completer = ConditionalCompleter(
    sql_completer,
    Condition(is_sql_mode)
)
3. 去重补全器
from prompt_toolkit.completion import DeduplicateCompleter

# 确保补全结果不重复
deduplicated_completer = DeduplicateCompleter(combined_completer)
4. 线程安全补全器

对于计算密集型补全操作,可以使用 ThreadedCompleter

from prompt_toolkit.completion import ThreadedCompleter

# 在后台线程中运行补全逻辑
threaded_completer = ThreadedCompleter(heavy_duty_completer)

补全事件处理

CompleteEvent 对象提供了补全触发上下文信息:

class SmartCompleter(Completer):
    def get_completions(self, document, complete_event):
        if complete_event.completion_requested:
            # 用户显式请求补全(按Tab键)
            yield from self.get_detailed_completions(document)
        elif complete_event.text_inserted:
            # 输入时自动补全
            yield from self.get_quick_completions(document)

性能优化技巧

  1. 延迟加载:对于大型数据集,使用生成器而非列表
  2. 缓存机制:对静态数据实施缓存策略
  3. 异步支持:实现 get_completions_async 方法
  4. 智能过滤:在数据源层面进行初步过滤
class OptimizedCompleter(Completer):
    def __init__(self):
        self._cache = {}
    
    def get_completions(self, document, complete_event):
        word = document.get_word_before_cursor()
        
        # 使用缓存避免重复计算
        if word not in self._cache:
            self._cache[word] = list(self._generate_completions(word))
        
        yield from self._cache[word]
    
    def _generate_completions(self, word):
        # 实际生成补全的逻辑
        for item in self.data_source:
            if item.startswith(word):
                yield Completion(item, start_position=-len(word))

实际应用示例

以下是一个完整的自定义补全器示例,模拟Git命令补全:

from prompt_toolkit.completion import Completer, Completion, NestedCompleter
from prompt_toolkit.document import Document

class GitCompleter(Completer):
    def __init__(self):
        self.commands = {
            'add': {
                '--all': None,
                '--patch': None,
                '--verbose': None,
            },
            'commit': {
                '--message': None,
                '--amend': None,
                '--all': None,
            },
            'push': {
                '--force': None,
                '--tags': None,
                'origin': {
                    'master': None,
                    'main': None,
                    'develop': None,
                }
            },
            'branch': {
                '-d': None,
                '-D': None,
                '--list': None,
            }
        }
    
    def get_completions(self, document, complete_event):
        text = document.text_before_cursor
        words = text.split()
        current_word = document.get_word_before_cursor()
        
        # 根据当前命令上下文提供补全
        if len(words) == 1:
            # 主命令补全
            for cmd in self.commands:
                if cmd.startswith(current_word):
                    yield Completion(cmd, start_position=-len(current_word))
        elif len(words) >= 2:
            # 子命令和选项补全
            main_cmd = words[0]
            if main_cmd in self.commands:
                options = self._get_options(self.commands[main_cmd])
                for opt in options:
                    if opt.startswith(current_word):
                        yield Completion(opt, start_position=-len(current_word))
    
    def _get_options(self, node, prefix=''):
        options = []
        if node is None:
            return options
        
        for key, value in node.items():
            full_key = f"{prefix}{key}"
            options.append(full_key)
            if isinstance(value, dict):
                options.extend(self._get_options(value, f"{full_key} "))
        return options

这个Git补全器能够根据用户输入的上下文智能地提供命令、选项和参数的补全建议,大大提升了命令行工具的使用体验。

通过Python Prompt Toolkit的自动补全机制,开发者可以创建出高度智能和用户友好的命令行界面,无论是简单的单词补全还是复杂的上下文感知补全,都能得到很好的支持。

输入验证与历史记录管理

在构建交互式命令行应用时,输入验证和历史记录管理是两个至关重要的功能。Python Prompt Toolkit 提供了强大而灵活的机制来处理这两个方面,让开发者能够创建既安全又用户友好的命令行界面。

输入验证机制

Python Prompt Toolkit 的输入验证系统基于 Validator 抽象基类,提供了多种验证策略和灵活的配置选项。

基本验证器使用

最简单的验证器可以通过 Validator.from_callable() 方法创建,该方法接受一个验证函数和错误消息:

from prompt_toolkit import prompt
from prompt_toolkit.validation import Validator

def is_valid_email(text):
    return "@" in text

validator = Validator.from_callable(
    is_valid_email,
    error_message="Not a valid e-mail address (Does not contain an @).",
    move_cursor_to_end=True,
)

# 验证仅在按下回车时进行
email = prompt("Enter email: ", validator=validator, validate_while_typing=False)

# 实时验证(输入时即时验证)
email = prompt("Enter email: ", validator=validator, validate_while_typing=True)
验证器类层次结构

Prompt Toolkit 提供了多种验证器实现,构成了一个完整的类层次结构:

mermaid

自定义验证器实现

对于更复杂的验证需求,可以创建自定义验证器类:

from prompt_toolkit.validation import Validator, ValidationError
from prompt_toolkit.document import Document

class NumberRangeValidator(Validator):
    def __init__(self, min_value: int, max_value: int):
        self.min_value = min_value
        self.max_value = max_value
        
    def validate(self, document: Document):
        text = document.text
        if not text.isdigit():
            raise ValidationError(
                cursor_position=0,
                message="Input must be a number"
            )
        
        value = int(text)
        if value < self.min_value or value > self.max_value:
            raise ValidationError(
                cursor_position=len(text),
                message=f"Number must be between {self.min_value} and {self.max_value}"
            )

# 使用自定义验证器
validator = NumberRangeValidator(1, 100)
number = prompt("Enter number (1-100): ", validator=validator)
异步验证支持

对于耗时的验证操作,Prompt Toolkit 支持异步验证:

from prompt_toolkit.validation import ThreadedValidator
import time

def slow_validation(text):
    # 模拟耗时操作
    time.sleep(2)
    return text.startswith("valid")

validator = ThreadedValidator(
    Validator.from_callable(
        slow_validation,
        error_message="Input must start with 'valid'"
    )
)

result = prompt("Enter text: ", validator=validator)

历史记录管理系统

历史记录管理是交互式应用的核心功能之一,Prompt Toolkit 提供了多种历史记录存储后端和灵活的配置选项。

历史记录类体系

mermaid

内存历史记录

最简单的历史记录实现,适用于临时会话:

from prompt_toolkit import PromptSession
from prompt_toolkit.history import InMemoryHistory

# 创建内存历史记录并预填充数据
history = InMemoryHistory()
history.append_string("import os")
history.append_string('print("hello")')
history.append_string('print("world")')

session = PromptSession(history=history)

while True:
    text = session.prompt("Say something: ")
    print(f"You said: {text}")
文件历史记录

持久化历史记录到文件,支持跨会话保存:

from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory

# 历史记录将保存到 .example-history-file 文件中
history = FileHistory(".example-history-file")
session = PromptSession(history=history)

while True:
    text = session.prompt("Say something: ")
    print(f"You said: {text}")

文件历史记录使用特定的格式存储数据,包括时间戳和多行支持:

# 2024-01-15 10:30:45.123456
+import os
# 2024-01-15 10:31:22.654321
+print("hello world")
+print("this is multiline")
异步历史记录加载

对于大型历史记录文件,可以使用线程化历史记录实现异步加载:

from prompt_toolkit import PromptSession
from prompt_toolkit.history import ThreadedHistory, FileHistory
import time

class SlowHistory(History):
    """模拟慢速历史记录加载"""
    def load_history_strings(self):
        for i in range(100):
            time.sleep(0.1)  # 模拟加载延迟
            yield f"history-item-{i}"
            
    def store_string(self, string):
        pass  # 不实际存储

# 使用线程化历史记录实现非阻塞加载
history = ThreadedHistory(SlowHistory())
session = PromptSession(history=history)

text = session.prompt("Input: ")
print(f"Received: {text}")

高级验证模式

条件验证

根据特定条件启用或禁用验证:

from prompt_toolkit.validation import ConditionalValidator
from prompt_toolkit.filters import Condition

def should_validate():
    # 复杂的条件逻辑
    return True

validator = ConditionalValidator(
    NumberRangeValidator(1, 100),
    filter=Condition(should_validate)
)
动态验证器

根据运行时状态动态选择验证器:

from prompt_toolkit.validation import DynamicValidator

def get_current_validator():
    if some_condition:
        return NumberRangeValidator(1, 50)
    else:
        return NumberRangeValidator(51, 100)

dynamic_validator = DynamicValidator(get_current_validator)
正则表达式语法验证

使用正则语言模块进行复杂的语法验证:

from prompt_toolkit.contrib.regular_languages import compile
from prompt_toolkit.contrib.regular_languages.validation import GrammarValidator

# 定义命令语法
grammar = compile(r"""
    command = (create|delete) \s+ entity
    create = 'create'
    delete = 'delete' 
    entity = ('user'|'group') \s+ name
    name = [a-zA-Z0-9_]+
""")

# 创建语法验证器
validator = GrammarValidator(grammar, validators={})

try:
    validator.validate(Document("create user john"))
    print("Valid command")
except ValidationError as e:
    print(f"Invalid command: {e.message}")

历史记录搜索与导航

Prompt Toolkit 提供了强大的历史记录搜索功能:

from prompt_toolkit import PromptSession
from prompt_toolkit.history import InMemoryHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory

history = InMemoryHistory()
history.append_string("import numpy as np")
history.append_string("import pandas as pd")
history.append_string("import matplotlib.pyplot as plt")

session = PromptSession(
    history=history,
    auto_suggest=AutoSuggestFromHistory(),
    enable_history_search=True
)

# 启用历史搜索后,用户可以:
# - 使用上箭头进行部分字符串匹配
# - 使用 Ctrl+R 进行反向搜索
# - 使用自动建议功能
text = session.prompt("Python import: ")

验证与历史记录的集成应用

将验证和历史记录结合使用,创建完整的交互式体验:

from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.validation import Validator, ValidationError
from prompt_toolkit.document import Document

class SQLValidator(Validator):
    def validate(self, document: Document):
        text = document.text.strip()
        if not text:
            return
            
        # 简单的SQL语法检查
        if not text.lower().startswith(('select', 'insert', 'update', 'delete')):
            raise ValidationError(
                cursor_position=0,
                message="SQL must start with SELECT, INSERT, UPDATE, or DELETE"
            )
            
        if not text.endswith(';'):
            raise ValidationError(
                cursor_position=len(text),
                message="SQL statement must end with semicolon"
            )

# 创建带验证的SQL历史记录会话
history = FileHistory(".sql-history")
validator = SQLValidator()

session = PromptSession(
    history=history,
    validator=validator,
    validate_while_typing=True,
    multiline=True
)

while True:
    try:
        query = session.prompt("SQL> ")
        print(f"Executing: {query}")
        # 执行SQL查询...
        
    except KeyboardInterrupt:
        continue
    except EOFError:
        break

性能优化与最佳实践

验证性能优化

对于复杂的验证逻辑,考虑使用异步验证:

from prompt_toolkit.validation import ThreadedValidator
import re

def complex_validation(text):
    # 复杂的正则表达式匹配
    pattern = r'^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}$'
    return re.match(pattern, text) is not None

# 使用线程化验证器避免界面卡顿
validator = ThreadedValidator(
    Validator.from_callable(
        complex_validation,
        error_message="Invalid email format"
    )
)
历史记录内存管理

对于大型历史记录,实施内存管理策略:

from collections import deque
from prompt_toolkit.history import History

class LimitedHistory(History):
    def __init__(self, max_size=1000):
        super().__init__()
        self._storage = deque(maxlen=max_size)
        
    def load_history_strings(self):
        return list(self._storage)
        
    def store_string(self, string):
        self._storage.append(string)
        
    def append_string(self, string):
        # 去重逻辑
        if string and (not self._storage or self._storage[-1] != string):
            self._storage.append(string)
            self.store_string(string)
验证状态管理

有效管理验证状态以提高性能:

from prompt_toolkit.validation import ValidationError
from prompt_toolkit.document import Document

class StatefulValidator(Validator):
    def __init__(self):
        self._last_text = None
        self._last_result = None
        
    def validate(self, document: Document):
        text = document.text
        
        # 缓存验证结果
        if text == self._last_text:
            if not self._last_result:
                raise ValidationError(
                    cursor_position=0,
                    message="Invalid input (cached)"
                )
            return
            
        # 执行实际验证
        is_valid = self._perform_validation(text)
        self._last_text = text
        self._last_result = is_valid
        
        if not is_valid:
            raise ValidationError(
                cursor_position=0,
                message="Invalid input"
            )
            
    def _perform_validation(self, text):
        # 实际的验证逻辑
        return len(text) > 3

通过合理使用 Python Prompt Toolkit 的输入验证和历史记录管理功能,开发者可以构建出既强大又用户友好的命令行应用程序。这些功能不仅提高了应用的健壮性,还显著改善了用户体验。

总结

Python Prompt Toolkit提供了构建现代化命令行应用的完整解决方案。从基础的PromptSession状态管理到高级的语法高亮、智能补全、输入验证和历史记录功能,该库展现了强大的灵活性和扩展性。通过合理配置会话参数、集成Pygments实现语法高亮、自定义补全逻辑、实施输入验证策略以及优化历史记录管理,开发者可以创建出既美观又功能强大的交互式命令行工具。这些功能的有机结合使得Prompt Toolkit成为Python生态中命令行界面开发的首选工具,能够满足从简单提示到复杂全屏编辑器的各种应用场景需求。

【免费下载链接】python-prompt-toolkit Library for building powerful interactive command line applications in Python 【免费下载链接】python-prompt-toolkit 项目地址: https://gitcode.com/gh_mirrors/py/python-prompt-toolkit

Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐