Skip to content

async_driver

scrapli_community.mikrotik.routeros.async_driver

AsyncMikrotikRouterOSDriver

Bases: AsyncGenericDriver

Source code in mikrotik/routeros/async_driver.py
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
class AsyncMikrotikRouterOSDriver(AsyncGenericDriver):
    def __init__(self, **kwargs: Any) -> None:
        """
        Mikrotik RouterOS platform class

        Args:
            kwargs: keyword args

        Returns:
            N/A

        Raises:
            N/A

        """

        # Append login options to the username according to
        # https://wiki.mikrotik.com/wiki/Manual:Console_login_process
        kwargs["auth_username"] += "+cet511w4098h"

        super().__init__(**kwargs)

    async def send_command(
        self,
        command: str,
        *,
        strip_prompt: bool = True,
        failed_when_contains: Optional[Union[str, List[str]]] = None,
        timeout_ops: Optional[float] = None,
    ) -> Response:
        """
        mikrotik_routeros send_command method

        Args:
            command: string to send to device in privilege exec mode
            strip_prompt: True/False strip prompt from returned output
            failed_when_contains: string or list of strings indicating failure if found in response
            timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
                the duration of the operation, value is reset to initial value after operation is
                completed

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """

        # RouterOS echoes the prompt/pattern twice after sending a command,
        # modify the prompt pattern accordingly to catch both echoes.
        #
        # [user@HOSTNAME]> /command\r\n[user@HOSTNAME]> /command\r\nOUTPUT...
        #
        old_comms_prompt_pattern = self.comms_prompt_pattern
        self.comms_prompt_pattern = f"{old_comms_prompt_pattern}.*{old_comms_prompt_pattern}"

        response = await super().send_command(
            command,
            strip_prompt=strip_prompt,
            failed_when_contains=failed_when_contains,
            timeout_ops=timeout_ops,
        )

        # Change the prompt pattern back to the original one.
        self.comms_prompt_pattern = old_comms_prompt_pattern

        # Since the command is echoed twice, and scrapli only removes it once, we need to
        # manually remove the second command echo by stripping the first line from the output.
        if response.result.count("\n") > 0 and command in response.result.split("\n")[0]:
            response.result = "\n".join(response.result.split("\n")[1:])

        return response

__init__(**kwargs: Any) -> None

Mikrotik RouterOS platform class

Parameters:

Name Type Description Default
kwargs

keyword args

required

Returns:

Type Description
None

N/A

Source code in mikrotik/routeros/async_driver.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(self, **kwargs: Any) -> None:
    """
    Mikrotik RouterOS platform class

    Args:
        kwargs: keyword args

    Returns:
        N/A

    Raises:
        N/A

    """

    # Append login options to the username according to
    # https://wiki.mikrotik.com/wiki/Manual:Console_login_process
    kwargs["auth_username"] += "+cet511w4098h"

    super().__init__(**kwargs)

send_command(command: str, *, strip_prompt: bool = True, failed_when_contains: Optional[Union[str, List[str]]] = None, timeout_ops: Optional[float] = None) -> Response async

mikrotik_routeros send_command method

Parameters:

Name Type Description Default
command str

string to send to device in privilege exec mode

required
strip_prompt bool

True/False strip prompt from returned output

True
failed_when_contains Optional[Union[str, List[str]]]

string or list of strings indicating failure if found in response

None
timeout_ops Optional[float]

timeout ops value for this operation; only sets the timeout_ops value for the duration of the operation, value is reset to initial value after operation is completed

None

Returns:

Name Type Description
Response Response

Scrapli Response object

Source code in mikrotik/routeros/async_driver.py
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
async def send_command(
    self,
    command: str,
    *,
    strip_prompt: bool = True,
    failed_when_contains: Optional[Union[str, List[str]]] = None,
    timeout_ops: Optional[float] = None,
) -> Response:
    """
    mikrotik_routeros send_command method

    Args:
        command: string to send to device in privilege exec mode
        strip_prompt: True/False strip prompt from returned output
        failed_when_contains: string or list of strings indicating failure if found in response
        timeout_ops: timeout ops value for this operation; only sets the timeout_ops value for
            the duration of the operation, value is reset to initial value after operation is
            completed

    Returns:
        Response: Scrapli Response object

    Raises:
        N/A

    """

    # RouterOS echoes the prompt/pattern twice after sending a command,
    # modify the prompt pattern accordingly to catch both echoes.
    #
    # [user@HOSTNAME]> /command\r\n[user@HOSTNAME]> /command\r\nOUTPUT...
    #
    old_comms_prompt_pattern = self.comms_prompt_pattern
    self.comms_prompt_pattern = f"{old_comms_prompt_pattern}.*{old_comms_prompt_pattern}"

    response = await super().send_command(
        command,
        strip_prompt=strip_prompt,
        failed_when_contains=failed_when_contains,
        timeout_ops=timeout_ops,
    )

    # Change the prompt pattern back to the original one.
    self.comms_prompt_pattern = old_comms_prompt_pattern

    # Since the command is echoed twice, and scrapli only removes it once, we need to
    # manually remove the second command echo by stripping the first line from the output.
    if response.result.count("\n") > 0 and command in response.result.split("\n")[0]:
        response.result = "\n".join(response.result.split("\n")[1:])

    return response

default_async_on_close(conn: AsyncGenericDriver) -> None async

Async mikrotik_routeros default on_close callable

Parameters:

Name Type Description Default
conn AsyncGenericDriver

AsyncGenericDriver object

required

Returns:

Type Description
None

N/A

Source code in mikrotik/routeros/async_driver.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
async def default_async_on_close(conn: AsyncGenericDriver) -> None:
    """
    Async mikrotik_routeros default on_close callable

    Args:
        conn: AsyncGenericDriver object

    Returns:
        N/A

    Raises:
        N/A

    """
    conn.channel.write(channel_input="/quit")
    conn.channel.send_return()