Python library to use the pseudo-tty of a docker container
npx @tessl/cli install tessl/pypi-dockerpty@0.4.0Python library to use the pseudo-tty of a docker container. dockerpty provides functionality to operate the pseudo-tty (PTY) allocated to a Docker container using the Python client, enabling developers to interact with Docker containers through a terminal interface by establishing a bridge between the host system's TTY and the container's PTY.
pip install dockerptysix >= 1.3.0 (docker-py assumed to be installed)import dockerptyFor advanced usage with specific classes:
from dockerpty import PseudoTerminal, RunOperation, ExecOperationimport docker
import dockerpty
# Create Docker client and container
client = docker.Client()
container = client.create_container(
image='busybox:latest',
stdin_open=True,
tty=True,
command='/bin/sh',
)
# Start PTY session - this will hijack the current terminal
dockerpty.start(client, container)Execute commands in existing containers:
import docker
import dockerpty
client = docker.Client()
container = 'container_id_or_name'
# Execute a command with PTY support
dockerpty.exec_command(client, container, '/bin/bash')dockerpty uses a multi-layered architecture for terminal management:
docker run (RunOperation) and docker exec (ExecOperation)The library handles complexities including raw mode configuration, non-blocking I/O operations, signal handling for window resizing (SIGWINCH), and proper resource cleanup.
High-level functions that provide the primary interface to dockerpty's functionality. These are convenience wrappers around the core PseudoTerminal class.
def start(client, container, interactive=True, stdout=None, stderr=None, stdin=None, logs=None):
"""Present the PTY of the container inside the current process."""
def exec_command(client, container, command, interactive=True, stdout=None, stderr=None, stdin=None):
"""Run provided command via exec API in provided container."""
def start_exec(client, exec_id, interactive=True, stdout=None, stderr=None, stdin=None):
"""Start an exec session using exec_id."""The core classes that handle pseudo-terminal allocation, management, and control. PseudoTerminal is the main class that coordinates terminal operations.
class PseudoTerminal:
def __init__(self, client, operation):
"""Initialize the PTY using the docker.Client instance and operation."""
def start(self, sockets=None):
"""Start PTY management with terminal hijacking."""
def resize(self, size=None):
"""Resize the container's PTY."""
class WINCHHandler:
def __init__(self, pty):
"""Initialize a new WINCH handler for the given PTY."""Operation classes that abstract the differences between running containers and executing commands in existing containers.
class RunOperation:
def __init__(self, client, container, interactive=True, stdout=None, stderr=None, stdin=None, logs=None):
"""Initialize the PTY for docker run-like operations."""
class ExecOperation:
def __init__(self, client, exec_id, interactive=True, stdout=None, stderr=None, stdin=None):
"""Initialize the PTY for docker exec-like operations."""
def exec_create(client, container, command, interactive=True):
"""Create exec instance for container."""Classes for handling I/O streams, multiplexing/demultiplexing, and data pumping between file descriptors.
class Stream:
def __init__(self, fd):
"""Initialize the Stream for the file descriptor."""
class Demuxer:
def __init__(self, stream):
"""Initialize a new Demuxer reading from stream."""
class Pump:
def __init__(self, from_stream, to_stream, wait_for_output=True, propagate_close=True):
"""Initialize a Pump with a Stream to read from and another to write to."""Low-level terminal management for handling raw mode, TTY size detection, and terminal attribute manipulation.
class Terminal:
def __init__(self, fd, raw=True):
"""Initialize a terminal for the tty with stdin attached to fd."""
def size(fd):
"""Return a tuple (rows,cols) representing the size of the TTY fd."""
def set_blocking(fd, blocking=True):
"""Set the given file-descriptor blocking or non-blocking."""
def select(read_streams, write_streams, timeout=0):
"""Select the streams ready for reading, and streams ready for writing."""# Error constants
class Stream:
ERRNO_RECOVERABLE = [errno.EINTR, errno.EDEADLK, errno.EWOULDBLOCK]