CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-dockerpty

Python library to use the pseudo-tty of a docker container

Pending
Overview
Eval results
Files

container-operations.mddocs/

Container Operations

Operation classes that abstract the differences between running containers and executing commands in existing containers. These classes implement the Operation interface and handle the specific details of Docker API interactions.

Capabilities

Run Operation

Handles docker run-like commands where you start a new container with PTY support.

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.
        
        Parameters:
        - client: Docker client instance
        - container: Container dict or ID
        - interactive: bool, whether to enable interactive mode (default: True)
        - stdout: file-like object for stdout (default: sys.stdout)
        - stderr: file-like object for stderr (default: sys.stderr)
        - stdin: file-like object for stdin (default: sys.stdin)
        - logs: int, whether to include logs (default: None, shows deprecation warning)
        """
    
    def start(self, sockets=None, **kwargs):
        """
        Present the PTY of the container inside the current process.
        
        This will take over the current process' TTY until the container's PTY is closed.
        If the container is not running, it will be started.
        
        Parameters:
        - sockets: tuple of (stdin_socket, stdout_socket, stderr_socket), optional
        - **kwargs: Additional arguments passed to client.start()
        
        Returns:
        List of Pump instances for stream management
        """
    
    def israw(self, **kwargs):
        """
        Returns True if the PTY should operate in raw mode.
        
        If the container was not started with tty=True, this will return False.
        
        Parameters:
        - **kwargs: Additional arguments (unused)
        
        Returns:
        bool - True if PTY should use raw mode
        """
    
    def sockets(self):
        """
        Returns a tuple of sockets connected to the pty (stdin,stdout,stderr).
        
        If any of the sockets are not attached in the container, None is
        returned in the tuple.
        
        Returns:
        tuple - (stdin_socket, stdout_socket, stderr_socket) or None for missing sockets
        """
    
    def resize(self, height, width, **kwargs):
        """
        Resize pty within container.
        
        Parameters:
        - height: int, terminal height in rows
        - width: int, terminal width in columns
        - **kwargs: Additional arguments (unused)
        
        Returns:
        None
        """

Usage example:

import docker
from dockerpty import RunOperation, PseudoTerminal

client = docker.Client()
container = client.create_container(
    image='alpine:latest',
    stdin_open=True,
    tty=True,
    command='/bin/sh',
)

# Create and start operation
operation = RunOperation(client, container, interactive=True)
pty = PseudoTerminal(client, operation)
pty.start()

Exec Operation

Handles docker exec-like commands where you execute a command in an existing running container.

class ExecOperation:
    def __init__(self, client, exec_id, interactive=True, stdout=None, stderr=None, stdin=None):
        """
        Initialize the PTY for docker exec-like operations.
        
        Parameters:
        - client: Docker client instance
        - exec_id: Exec instance ID (from client.exec_create())
        - interactive: bool, whether to enable interactive mode (default: True)
        - stdout: file-like object for stdout (default: sys.stdout)
        - stderr: file-like object for stderr (default: sys.stderr)
        - stdin: file-like object for stdin (default: sys.stdin)
        """
    
    def start(self, sockets=None, **kwargs):
        """
        Start execution of the exec command.
        
        Parameters:
        - sockets: socket for exec I/O, optional
        - **kwargs: Additional arguments (unused)
        
        Returns:
        List of Pump instances for stream management
        """
    
    def israw(self, **kwargs):
        """
        Returns True if the PTY should operate in raw mode.
        
        If the exec was not started with tty=True, this will return False.
        
        Parameters:
        - **kwargs: Additional arguments (unused)
        
        Returns:
        bool - True if PTY should use raw mode
        """
    
    def sockets(self):
        """
        Return a single socket which is processing all I/O to exec.
        
        Returns:
        Stream or Demuxer instance for exec I/O
        """
    
    def resize(self, height, width, **kwargs):
        """
        Resize pty of an execed process.
        
        Parameters:
        - height: int, terminal height in rows
        - width: int, terminal width in columns
        - **kwargs: Additional arguments (unused)
        
        Returns:
        None
        """
    
    def is_process_tty(self):
        """
        Does execed process have allocated tty?
        
        Returns:
        bool - True if process has TTY allocated
        """

Usage example:

import docker
from dockerpty import ExecOperation, PseudoTerminal

client = docker.Client()
container_id = 'running_container_id'

# Create exec instance
exec_id = client.exec_create(container_id, '/bin/bash', tty=True, stdin=True)

# Create and start operation
operation = ExecOperation(client, exec_id['Id'], interactive=True)
pty = PseudoTerminal(client, operation)
pty.start()

Exec Creation Utility

Convenience function for creating exec instances.

def exec_create(client, container, command, interactive=True):
    """
    Create exec instance for container.
    
    Parameters:
    - client: Docker client instance
    - container: Container dict or ID
    - command: str or list, command to execute
    - interactive: bool, whether to enable interactive mode (default: True)
    
    Returns:
    dict - Exec instance information with 'Id' key
    """

Usage example:

import docker
from dockerpty import exec_create

client = docker.Client()
container_id = 'running_container_id'

# Create exec instance
exec_id = exec_create(client, container_id, ['/bin/bash', '-l'])
print(exec_id['Id'])  # Exec instance ID

Operation Base Class

Abstract base class defining the operation interface.

class Operation:
    def israw(self, **kwargs):
        """Are we dealing with a tty or not?"""
        raise NotImplementedError()
    
    def start(self, **kwargs):
        """Start execution."""
        raise NotImplementedError()
    
    def resize(self, height, width, **kwargs):
        """If we have terminal, resize it."""
        raise NotImplementedError()
    
    def sockets(self):
        """Return sockets for streams."""
        raise NotImplementedError()

Key Differences

RunOperation vs ExecOperation

RunOperation (for new containers):

  • Can start stopped containers
  • Returns separate stdin, stdout, stderr sockets
  • Handles container lifecycle management
  • Supports logs parameter for including container logs

ExecOperation (for existing containers):

  • Works only with running containers
  • Returns single multiplexed socket for all I/O
  • Cannot distinguish stderr from stdout in current implementation
  • More lightweight for command execution

Stream Handling

  • RunOperation: Uses separate streams that may be demultiplexed if no TTY
  • ExecOperation: Uses single stream, always demultiplexed if no TTY
  • Both support TTY mode for raw terminal interaction

Error Handling

  • Handles container inspection failures
  • Manages resize failures (container may have exited)
  • Supports graceful degradation when TTY is not available
  • Properly handles Docker API connectivity issues

Install with Tessl CLI

npx tessl i tessl/pypi-dockerpty

docs

container-operations.md

core-pty-management.md

index.md

main-entry-points.md

stream-management.md

terminal-control.md

tile.json