Skip to content

base_transport

scrapli.transport.base_transport

BaseTransport

Bases: ABC

Source code in transport/base/base_transport.py 
 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class BaseTransport(ABC):
    def __init__(self, base_transport_args: BaseTransportArgs) -> None:
        """
        Scrapli's transport base class

        Args:
            base_transport_args: base transport args dataclass

        Returns:
            None

        Raises:
            N/A

        """
        self._base_transport_args = base_transport_args

        self.logger = get_instance_logger(
            instance_name="scrapli.transport",
            host=self._base_transport_args.host,
            port=self._base_transport_args.port,
            uid=self._base_transport_args.logging_uid,
        )

    @abstractmethod
    def close(self) -> None:
        """
        Close the transport session

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """

    @abstractmethod
    def write(self, channel_input: bytes) -> None:
        """
        Write bytes into the transport session

        Args:
            channel_input: bytes to write to transport session

        Returns:
            None

        Raises:
            N/A

        """

    @abstractmethod
    def isalive(self) -> bool:
        """
        Check if transport is alive

        Args:
            N/A

        Returns:
            bool: True/False if transport is alive

        Raises:
            N/A

        """

    def _pre_open_closing_log(self, closing: bool = False) -> None:
        """
        Emit "pre open" log message for consistency between transports

        Args:
            closing: bool indicating if message is for closing not opening

        Returns:
            None

        Raises:
            N/A

        """
        operation = "closing" if closing else "opening"

        self.logger.debug(
            f"{operation} transport connection to '{self._base_transport_args.host}' on port "
            f"'{self._base_transport_args.port}'"
        )

    def _post_open_closing_log(self, closing: bool = False) -> None:
        """
        Emit "post open" log message for consistency between transports

        Args:
            closing: bool indicating if message is for closing not opening

        Returns:
            None

        Raises:
            N/A

        """
        operation = "closed" if closing else "opened"

        self.logger.debug(
            f"transport connection to '{self._base_transport_args.host}' on port "
            f"'{self._base_transport_args.port}' {operation} successfully"
        )

__init__(base_transport_args: BaseTransportArgs) -> None

Scrapli's transport base class

Parameters:

Name Type Description Default
base_transport_args BaseTransportArgs

base transport args dataclass

 required

Returns:

Type Description
None

None
Source code in transport/base/base_transport.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(self, base_transport_args: BaseTransportArgs) -> None:
    """
    Scrapli's transport base class

    Args:
        base_transport_args: base transport args dataclass

    Returns:
        None

    Raises:
        N/A

    """
    self._base_transport_args = base_transport_args

    self.logger = get_instance_logger(
        instance_name="scrapli.transport",
        host=self._base_transport_args.host,
        port=self._base_transport_args.port,
        uid=self._base_transport_args.logging_uid,
    )

close() -> None abstractmethod

Close the transport session

Returns:

Type Description
None

None
Source code in transport/base/base_transport.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@abstractmethod
def close(self) -> None:
    """
    Close the transport session

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """

isalive() -> bool abstractmethod

Check if transport is alive

Returns:

Name Type Description
bool bool

True/False if transport is alive
Source code in transport/base/base_transport.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@abstractmethod
def isalive(self) -> bool:
    """
    Check if transport is alive

    Args:
        N/A

    Returns:
        bool: True/False if transport is alive

    Raises:
        N/A

    """

write(channel_input: bytes) -> None abstractmethod

Write bytes into the transport session

Parameters:

Name Type Description Default
channel_input bytes

bytes to write to transport session

 required

Returns:

Type Description
None

None
Source code in transport/base/base_transport.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@abstractmethod
def write(self, channel_input: bytes) -> None:
    """
    Write bytes into the transport session

    Args:
        channel_input: bytes to write to transport session

    Returns:
        None

    Raises:
        N/A

    """