1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
#!/usr/bin/env python
"""
Example of running a prompt_toolkit application in an asyncssh server.
"""
import asyncio
import logging
import asyncssh
from pygments.lexers.html import HtmlLexer
from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.contrib.ssh import PromptToolkitSSHServer, PromptToolkitSSHSession
from prompt_toolkit.lexers import PygmentsLexer
from prompt_toolkit.shortcuts import ProgressBar, print_formatted_text
from prompt_toolkit.shortcuts.dialogs import input_dialog, yes_no_dialog
from prompt_toolkit.shortcuts.prompt import PromptSession
animal_completer = WordCompleter(
[
"alligator",
"ant",
"ape",
"bat",
"bear",
"beaver",
"bee",
"bison",
"butterfly",
"cat",
"chicken",
"crocodile",
"dinosaur",
"dog",
"dolphin",
"dove",
"duck",
"eagle",
"elephant",
"fish",
"goat",
"gorilla",
"kangaroo",
"leopard",
"lion",
"mouse",
"rabbit",
"rat",
"snake",
"spider",
"turkey",
"turtle",
],
ignore_case=True,
)
async def interact(ssh_session: PromptToolkitSSHSession) -> None:
"""
The application interaction.
This will run automatically in a prompt_toolkit AppSession, which means
that any prompt_toolkit application (dialogs, prompts, etc...) will use the
SSH channel for input and output.
"""
prompt_session = PromptSession()
# Alias 'print_formatted_text', so that 'print' calls go to the SSH client.
print = print_formatted_text
print("We will be running a few prompt_toolkit applications through this ")
print("SSH connection.\n")
# Simple progress bar.
with ProgressBar() as pb:
for i in pb(range(50)):
await asyncio.sleep(0.1)
# Normal prompt.
text = await prompt_session.prompt_async("(normal prompt) Type something: ")
print("You typed", text)
# Prompt with auto completion.
text = await prompt_session.prompt_async(
"(autocompletion) Type an animal: ", completer=animal_completer
)
print("You typed", text)
# prompt with syntax highlighting.
text = await prompt_session.prompt_async(
"(HTML syntax highlighting) Type something: ", lexer=PygmentsLexer(HtmlLexer)
)
print("You typed", text)
# Show yes/no dialog.
await prompt_session.prompt_async("Showing yes/no dialog... [ENTER]")
await yes_no_dialog("Yes/no dialog", "Running over asyncssh").run_async()
# Show input dialog
await prompt_session.prompt_async("Showing input dialog... [ENTER]")
await input_dialog("Input dialog", "Running over asyncssh").run_async()
async def main(port=8222):
# Set up logging.
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
await asyncssh.create_server(
lambda: PromptToolkitSSHServer(interact),
"",
port,
server_host_keys=["/etc/ssh/ssh_host_ecdsa_key"],
)
# Run forever.
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())
|
