Skip to content

Python

PYTHON_INTERPRETER

This command will spawn a interactive python interpreter in a child process. Any text/command that is entered is sent to the stdin of the child, and the result is read from stdout and stderr. It is then written to a RichLog. This should meet any basic needs but your mileage may vary.

Python

Bases: Command

Command for spawning an interactive python interpreter.

Source code in src/textual_shell/commands/python.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class Python(Command):
    """Command for spawning an interactive python interpreter."""
    DEFINITION = {
        'python': CommandNode(
            name='python',
            description='Spawn a Python Interpreter'
        )
    }

    def create_job(self, *args):
        """Creates the job to spawn the interpreter."""
        return RunPythonInterpreter(
            shell=self.shell,
            cmd=self.name
        )

create_job(*args)

Creates the job to spawn the interpreter.

Source code in src/textual_shell/commands/python.py 
239
240
241
242
243
244
def create_job(self, *args):
    """Creates the job to spawn the interpreter."""
    return RunPythonInterpreter(
        shell=self.shell,
        cmd=self.name
    )

PythonArea

Bases: ShellArea

Custom TextArea to replicate a python interpreter's interface.

Source code in src/textual_shell/commands/python.py 
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
class PythonArea(ShellArea):
    """Custom TextArea to replicate a 
    python interpreter's interface."""
    BINDINGS = [
        Binding('tab', 'tab', 'Insert Tab', show=False, priority=True)
    ]

    prompt = '>>> '
    multiline_char = ':'
    multiline_prompt = '\n... '

    def action_tab(self):
        """Insert tab"""
        self.insert('\t')

    def send_command(self, text: str):
        """
        Post the Execute message.

        Args:
            text (str): The command to post.
        """
        text = text[len(self.prompt):]
        self.post_message(self.Execute(text))

        self.current_history_index = None
        self.action_clear()
        self.action_cursor_line_end()
        self.multiline = False

    def action_enter_pressed(self):
        """Handler for pressing the enter key."""
        self.action_cursor_line_end()

        text = self.text
        if text.endswith(self.multiline_char) and self.multiline == False:
            self.insert(self.multiline_prompt)
            self.multiline = True

        elif self.multiline == True:
            if text.endswith(self.multiline_prompt):
                self.send_command(text)

            else:
                self.insert(self.multiline_prompt)

        else:
            self.send_command(text)

action_enter_pressed()

Handler for pressing the enter key.

Source code in src/textual_shell/commands/python.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def action_enter_pressed(self):
    """Handler for pressing the enter key."""
    self.action_cursor_line_end()

    text = self.text
    if text.endswith(self.multiline_char) and self.multiline == False:
        self.insert(self.multiline_prompt)
        self.multiline = True

    elif self.multiline == True:
        if text.endswith(self.multiline_prompt):
            self.send_command(text)

        else:
            self.insert(self.multiline_prompt)

    else:
        self.send_command(text)

action_tab()

Insert tab

Source code in src/textual_shell/commands/python.py 
25
26
27
def action_tab(self):
    """Insert tab"""
    self.insert('\t')

send_command(text)

Post the Execute message.

Parameters:

Name Type Description Default
text str

The command to post.

 required
Source code in src/textual_shell/commands/python.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def send_command(self, text: str):
    """
    Post the Execute message.

    Args:
        text (str): The command to post.
    """
    text = text[len(self.prompt):]
    self.post_message(self.Execute(text))

    self.current_history_index = None
    self.action_clear()
    self.action_cursor_line_end()
    self.multiline = False

PythonInterpreter

Bases: Screen

Screen to render an interactive python interpreter.

Parameters:

Name Type Description Default
task Task

The asyncio task of the job.

 required
Source code in src/textual_shell/commands/python.py 
 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class PythonInterpreter(Screen):
    """
    Screen to render an interactive python interpreter.

    Args:
        task (asyncio.Task): The asyncio task of the job.
    """

    BINDINGS = [
        Binding('ctrl+z', 'background_job', 'Background the job.', priority=True),
        Binding('ctrl+d', 'kill_interpreter', 'Close the shell', priority=True),
    ]

    DEFAULT_CSS = """
        RichLog {
            height: auto;
            padding-left: 1;
            max-height: 90%;
            border: hidden;
            background: transparent;
        }

        TextArea {
            height: auto;
            border: hidden;
            background: transparent;
        }

        TextArea:focus {
            border: none;
        }
    """

    def __init__(
        self,
        task: Annotated[asyncio.Task, 'The asyncio task of the job the shell is running in.'],
        *args, **kwargs
    ) -> None:
        super().__init__(*args, **kwargs)
        self.prompt = '>>> '
        self.interpreter_task = task
        self.run_worker(self.setup())

    def compose(self) -> ComposeResult:
        yield RichLog(markup=True, wrap=True)
        yield PythonArea()

    def on_mount(self) -> None:
        text_area = self.query_one(PythonArea)
        text_area.focus()

    def action_background_job(self) -> None:
        """Background the interpreter and 
        return to the main screen."""
        self.app.pop_screen()

    def action_kill_interpreter(self) -> None:
        """Kill the interpreter and 
        return to the main screen"""
        for task in self.tasks:
            task.cancel()

        self.PYTHON_INTERPRETER.kill()
        self.interpreter_task.cancel()
        self.app.pop_screen()

    async def setup(self):
        """Spawn the child process to run the python interpreter.
        Also create the tasks for reading stdout and stderr."""
        self.PYTHON_INTERPRETER = await asyncio.create_subprocess_exec(
            'python',
            '-i',
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout_task = asyncio.create_task(
            self.read_stdout(),
            name='stdout_task'
        )

        stderr_task = asyncio.create_task(
            self.read_stderr(),
            name='stderr_task'
        )

        self.tasks = [stdout_task, stderr_task]

    async def on_shell_area_execute(
        self,
        event: ShellArea.Execute
    ) -> None:
        """
        Execute the python code.

        Args:
            event (ShellArea.Execute): The message with the code.
        """
        rich_log = self.query_one(RichLog)
        interpreter = self.query_one(PythonArea)

        text = event.command

        text = text.replace('\n... ', '\n')

        if text != '':
            interpreter.history_list.appendleft(text)

        if text == 'exit()':
            self.action_kill_shell()

        self.PYTHON_INTERPRETER.stdin.write(text.encode() + b'\n')
        await self.PYTHON_INTERPRETER.stdin.drain()

        rich_log.write(self.prompt + event.command)


    async def update_from_stdout(self, output) -> None:
        """Take stdout and write it to the RichLog."""
        rich_log = self.query_one(RichLog)
        rich_log.write(output)

    async def update_from_stderr(self, error) -> None:
        """Take from stderr and write it to the RichLog."""
        rich_log = self.query_one(RichLog)
        rich_log.write(error)

    async def read_stdout(self):
        """Coroutine for reading stdout and updating the RichLog."""
        try:
            async for line in self.PYTHON_INTERPRETER.stdout:
                decoded = line.decode().strip()
                await self.update_from_stdout(decoded)

        except asyncio.CancelledError:
            return

    async def read_stderr(self):
        """Coroutine for reading stderr and updating the RichLog."""
        try:
            async for line in self.PYTHON_INTERPRETER.stderr:
                decoded = line.decode().strip()
                await self.update_from_stderr(decoded)

        except asyncio.CancelledError:
            return

action_background_job()

Background the interpreter and return to the main screen.

Source code in src/textual_shell/commands/python.py 
115
116
117
118
def action_background_job(self) -> None:
    """Background the interpreter and 
    return to the main screen."""
    self.app.pop_screen()

action_kill_interpreter()

Kill the interpreter and return to the main screen

Source code in src/textual_shell/commands/python.py 
120
121
122
123
124
125
126
127
128
def action_kill_interpreter(self) -> None:
    """Kill the interpreter and 
    return to the main screen"""
    for task in self.tasks:
        task.cancel()

    self.PYTHON_INTERPRETER.kill()
    self.interpreter_task.cancel()
    self.app.pop_screen()

on_shell_area_execute(event) async

Execute the python code.

Parameters:

Name Type Description Default
event Execute

The message with the code.

 required
Source code in src/textual_shell/commands/python.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
async def on_shell_area_execute(
    self,
    event: ShellArea.Execute
) -> None:
    """
    Execute the python code.

    Args:
        event (ShellArea.Execute): The message with the code.
    """
    rich_log = self.query_one(RichLog)
    interpreter = self.query_one(PythonArea)

    text = event.command

    text = text.replace('\n... ', '\n')

    if text != '':
        interpreter.history_list.appendleft(text)

    if text == 'exit()':
        self.action_kill_shell()

    self.PYTHON_INTERPRETER.stdin.write(text.encode() + b'\n')
    await self.PYTHON_INTERPRETER.stdin.drain()

    rich_log.write(self.prompt + event.command)

read_stderr() async

Coroutine for reading stderr and updating the RichLog.

Source code in src/textual_shell/commands/python.py 
201
202
203
204
205
206
207
208
209
async def read_stderr(self):
    """Coroutine for reading stderr and updating the RichLog."""
    try:
        async for line in self.PYTHON_INTERPRETER.stderr:
            decoded = line.decode().strip()
            await self.update_from_stderr(decoded)

    except asyncio.CancelledError:
        return

read_stdout() async

Coroutine for reading stdout and updating the RichLog.

Source code in src/textual_shell/commands/python.py 
191
192
193
194
195
196
197
198
199
async def read_stdout(self):
    """Coroutine for reading stdout and updating the RichLog."""
    try:
        async for line in self.PYTHON_INTERPRETER.stdout:
            decoded = line.decode().strip()
            await self.update_from_stdout(decoded)

    except asyncio.CancelledError:
        return

setup() async

Spawn the child process to run the python interpreter. Also create the tasks for reading stdout and stderr.

Source code in src/textual_shell/commands/python.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
async def setup(self):
    """Spawn the child process to run the python interpreter.
    Also create the tasks for reading stdout and stderr."""
    self.PYTHON_INTERPRETER = await asyncio.create_subprocess_exec(
        'python',
        '-i',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout_task = asyncio.create_task(
        self.read_stdout(),
        name='stdout_task'
    )

    stderr_task = asyncio.create_task(
        self.read_stderr(),
        name='stderr_task'
    )

    self.tasks = [stdout_task, stderr_task]

update_from_stderr(error) async

Take from stderr and write it to the RichLog.

Source code in src/textual_shell/commands/python.py 
186
187
188
189
async def update_from_stderr(self, error) -> None:
    """Take from stderr and write it to the RichLog."""
    rich_log = self.query_one(RichLog)
    rich_log.write(error)

update_from_stdout(output) async

Take stdout and write it to the RichLog.

Source code in src/textual_shell/commands/python.py 
181
182
183
184
async def update_from_stdout(self, output) -> None:
    """Take stdout and write it to the RichLog."""
    rich_log = self.query_one(RichLog)
    rich_log.write(output)

RunPythonInterpreter

Bases: Job

Creates the python interpreter screen and installs it onto the screen stack

Source code in src/textual_shell/commands/python.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class RunPythonInterpreter(Job):
    """Creates the python interpreter screen and
      installs it onto the screen stack"""

    async def execute(self):
        """Execute the interpreter."""
        self.running()

        self.screen = PythonInterpreter(self.task)
        self.shell.app.install_screen(self.screen, name=self.id)
        self.shell.app.push_screen(self.screen)

        await self.wait_for_cancel()

        self.shell.app.uninstall_screen(self.screen)
        self.completed()

execute() async

Execute the interpreter.

Source code in src/textual_shell/commands/python.py 
216
217
218
219
220
221
222
223
224
225
226
227
async def execute(self):
    """Execute the interpreter."""
    self.running()

    self.screen = PythonInterpreter(self.task)
    self.shell.app.install_screen(self.screen, name=self.id)
    self.shell.app.push_screen(self.screen)

    await self.wait_for_cancel()

    self.shell.app.uninstall_screen(self.screen)
    self.completed()