Module netmiko.channel

Classes

class Channel (*args: Any, **kwargs: Any)

Expand source code

class Channel(ABC):
    @abstractmethod
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Create the object."""
        pass

    # @abstractmethod
    # def __repr__(self) -> str:
    #     """String representation of the object."""
    #     pass
    #
    # @abstractmethod
    # def open(self, width: int = 511, height: int = 1000) -> None:
    #     """Create the underlying connection."""
    #     pass
    #
    # @abstractmethod
    # def close(self) -> None:
    #     """Close the underlying connection."""
    #     pass
    #
    # @abstractmethod
    # def login(self) -> None:
    #     """Handle the channel login process for any channel that requires it."""
    #     pass

    @abstractmethod
    def read_buffer(self) -> str:
        """Single read of available data."""
        pass

    @abstractmethod
    def read_channel(self) -> str:
        """Read all of the available data from the channel."""
        pass

    @abstractmethod
    def write_channel(self, out_data: str) -> None:
        """Write data down the channel."""
        pass

    # @abstractmethod
    # def is_alive(self) -> bool:
    #     """Is the channel alive."""
    #     pass

Helper class that provides a standard way to create an ABC using inheritance.

Create the object.

Ancestors

• abc.ABC

Subclasses

• SSHChannel
• SerialChannel
• TelnetChannel

Methods

def read_buffer(self) ‑> str

Expand source code

@abstractmethod
def read_buffer(self) -> str:
    """Single read of available data."""
    pass

Single read of available data.

def read_channel(self) ‑> str

Expand source code

@abstractmethod
def read_channel(self) -> str:
    """Read all of the available data from the channel."""
    pass

Read all of the available data from the channel.

def write_channel(self, out_data: str) ‑> None

Expand source code

@abstractmethod
def write_channel(self, out_data: str) -> None:
    """Write data down the channel."""
    pass

Write data down the channel.

class SSHChannel (conn: paramiko.channel.Channel | None, encoding: str)

Expand source code

class SSHChannel(Channel):
    def __init__(self, conn: Optional[paramiko.Channel], encoding: str) -> None:
        """
        Placeholder __init__ method so that reading and writing can be moved to the
        channel class.
        """
        self.remote_conn = conn
        # FIX: move encoding to GlobalState object?
        self.encoding = encoding

    def write_channel(self, out_data: str) -> None:
        if self.remote_conn is None:
            raise WriteException("Attempt to write data, but there is no active channel.")
        self.remote_conn.sendall(write_bytes(out_data, encoding=self.encoding))

    def read_buffer(self) -> str:
        """Single read of available data."""
        if self.remote_conn is None:
            raise ReadException("Attempt to read, but there is no active channel.")
        output = ""
        if self.remote_conn.recv_ready():
            outbuf = self.remote_conn.recv(MAX_BUFFER)
            if len(outbuf) == 0:
                raise ReadException("Channel stream closed by remote device.")
            output += outbuf.decode(self.encoding, "ignore")
        return output

    def read_channel(self) -> str:
        """Read all of the available data from the channel."""
        if self.remote_conn is None:
            raise ReadException("Attempt to read, but there is no active channel.")
        output = ""
        while True:
            new_output = self.read_buffer()
            output += new_output
            if new_output == "":
                break
        return output

Helper class that provides a standard way to create an ABC using inheritance.

Placeholder init method so that reading and writing can be moved to the channel class.

Ancestors

• Channel
• abc.ABC

Inherited members

• Channel:
  • read_buffer
  • read_channel
  • write_channel

class SerialChannel (conn: serial.serialposix.Serial | None, encoding: str)

Expand source code

class SerialChannel(Channel):
    def __init__(self, conn: Optional[serial.Serial], encoding: str) -> None:
        """
        Placeholder __init__ method so that reading and writing can be moved to the
        channel class.
        """
        self.remote_conn = conn
        # FIX: move encoding to GlobalState object?
        self.encoding = encoding

    def write_channel(self, out_data: str) -> None:
        if self.remote_conn is None:
            raise WriteException("Attempt to write data, but there is no active channel.")
        self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
        self.remote_conn.flush()

    def read_buffer(self) -> str:
        """Single read of available data."""
        if self.remote_conn is None:
            raise ReadException("Attempt to read, but there is no active channel.")
        if self.remote_conn.in_waiting > 0:
            output = self.remote_conn.read(self.remote_conn.in_waiting).decode(
                self.encoding, "ignore"
            )
            assert isinstance(output, str)
            return output
        else:
            return ""

    def read_channel(self) -> str:
        """Read all of the available data from the channel."""
        if self.remote_conn is None:
            raise ReadException("Attempt to read, but there is no active channel.")
        output = ""
        while self.remote_conn.in_waiting > 0:
            output += self.read_buffer()
        return output

Helper class that provides a standard way to create an ABC using inheritance.

Placeholder init method so that reading and writing can be moved to the channel class.

Ancestors

• Channel
• abc.ABC

Inherited members

• Channel:
  • read_buffer
  • read_channel
  • write_channel

class TelnetChannel (conn: netmiko._telnetlib.telnetlib.Telnet | None, encoding: str)

Expand source code

class TelnetChannel(Channel):
    def __init__(self, conn: Optional[telnetlib.Telnet], encoding: str) -> None:
        """
        Placeholder __init__ method so that reading and writing can be moved to the
        channel class.
        """
        self.remote_conn = conn
        # FIX: move encoding to GlobalState object?
        self.encoding = encoding

    def write_channel(self, out_data: str) -> None:
        if self.remote_conn is None:
            raise WriteException("Attempt to write data, but there is no active channel.")
        self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))  # type: ignore

    def read_buffer(self) -> str:
        """Single read of available data."""
        raise NotImplementedError

    def read_channel(self) -> str:
        """Read all of the available data from the channel.

        This recreates telnetlib's read_very_eager() behavior ("drain everything
        currently available, without blocking") but as an O(n) operation.

        read_very_eager() accumulates cooked data into a single immutable bytes
        buffer (cookedq) that it re-copies on every loop iteration, which is
        O(n^2) for large reads (see netmiko GH #3872). Instead we loop the
        public read_eager() primitive, which drains cookedq on every call, and
        accumulate its results in a Python list, joining once at the end. Each
        concat therefore stays O(chunk) and the whole read is O(n).

        read_eager() returns as soon as it has any cooked data, so looping it
        until it yields nothing reproduces read_very_eager()'s greedy draining.
        The EOFError contract (raised on a closed connection with no buffered
        data, and relied on by telnet_login) is preserved: it is re-raised only
        when no data was collected.
        """
        if self.remote_conn is None:
            raise ReadException("Attempt to read, but there is no active channel.")

        parts: list[bytes] = []
        while True:
            try:
                chunk = self.remote_conn.read_eager()  # type: ignore
            except EOFError:
                # Preserve read_very_eager()'s contract: only surface EOF when
                # there is no data to return.
                if not parts:
                    raise
                break
            if not chunk:
                break
            parts.append(chunk)
        return b"".join(parts).decode(self.encoding, "ignore")

Helper class that provides a standard way to create an ABC using inheritance.

Placeholder init method so that reading and writing can be moved to the channel class.

Ancestors

• Channel
• abc.ABC

Methods

def read_channel(self) ‑> str

Expand source code

def read_channel(self) -> str:
    """Read all of the available data from the channel.

    This recreates telnetlib's read_very_eager() behavior ("drain everything
    currently available, without blocking") but as an O(n) operation.

    read_very_eager() accumulates cooked data into a single immutable bytes
    buffer (cookedq) that it re-copies on every loop iteration, which is
    O(n^2) for large reads (see netmiko GH #3872). Instead we loop the
    public read_eager() primitive, which drains cookedq on every call, and
    accumulate its results in a Python list, joining once at the end. Each
    concat therefore stays O(chunk) and the whole read is O(n).

    read_eager() returns as soon as it has any cooked data, so looping it
    until it yields nothing reproduces read_very_eager()'s greedy draining.
    The EOFError contract (raised on a closed connection with no buffered
    data, and relied on by telnet_login) is preserved: it is re-raised only
    when no data was collected.
    """
    if self.remote_conn is None:
        raise ReadException("Attempt to read, but there is no active channel.")

    parts: list[bytes] = []
    while True:
        try:
            chunk = self.remote_conn.read_eager()  # type: ignore
        except EOFError:
            # Preserve read_very_eager()'s contract: only surface EOF when
            # there is no data to return.
            if not parts:
                raise
            break
        if not chunk:
            break
        parts.append(chunk)
    return b"".join(parts).decode(self.encoding, "ignore")

Read all of the available data from the channel.

This recreates telnetlib's read_very_eager() behavior ("drain everything currently available, without blocking") but as an O(n) operation.

read_very_eager() accumulates cooked data into a single immutable bytes buffer (cookedq) that it re-copies on every loop iteration, which is O(n^2) for large reads (see netmiko GH #3872). Instead we loop the public read_eager() primitive, which drains cookedq on every call, and accumulate its results in a Python list, joining once at the end. Each concat therefore stays O(chunk) and the whole read is O(n).

read_eager() returns as soon as it has any cooked data, so looping it until it yields nothing reproduces read_very_eager()'s greedy draining. The EOFError contract (raised on a closed connection with no buffered data, and relied on by telnet_login) is preserved: it is re-raised only when no data was collected.

Inherited members

• Channel:
  • read_buffer
  • write_channel