Skip to content

Module scrapli.driver.generic.base_driver

scrapli.driver.generic.base_driver

Expand source code
        
"""scrapli.driver.generic.base_driver"""
from typing import List, Optional, Tuple, Union

from scrapli.exceptions import ScrapliTypeError
from scrapli.helper import resolve_file
from scrapli.response import MultiResponse, Response


class BaseGenericDriver:
    @staticmethod
    def _pre_send_command(
        host: str, command: str, failed_when_contains: Optional[Union[str, List[str]]] = None
    ) -> Response:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            command: string to send to device in privilege exec mode
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(command, str):
            raise ScrapliTypeError(
                f"`send_command` expects a single string, got {type(command)}, "
                "to send a list of commands use the `send_commands` method instead."
            )

        response = Response(
            host=host,
            channel_input=command,
            failed_when_contains=failed_when_contains,
        )

        return response

    @staticmethod
    def _post_send_command(
        raw_response: bytes, processed_response: bytes, response: Response
    ) -> Response:
        """
        Handle post "send_command" tasks for consistency between sync/async versions

        Args:
            raw_response: raw response returned from the channel
            processed_response: processed response returned from the channel
            response: response object to update with channel results

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        response.record_response(result=processed_response)
        response.raw_result = raw_response
        return response

    @staticmethod
    def _pre_send_commands(commands: List[str]) -> MultiResponse:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            commands: list of strings to send to device in privilege exec mode

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(commands, list):
            raise ScrapliTypeError(
                f"`send_commands` expects a list of strings, got {type(commands)}, "
                "to send a single command use the `send_command` method instead."
            )

        responses = MultiResponse()

        return responses

    @staticmethod
    def _pre_send_from_file(file: str, caller: str) -> List[str]:
        """
        Handle pre "send_*_from_file" tasks for consistency between sync/async versions

        Args:
            file: string path to file
            caller: name of the calling method for more helpful error message

        Returns:
            list: list of commands/configs read from file

        Raises:
            ScrapliTypeError: if anything but a string is provided for `file`

        """
        if not isinstance(file, str):
            raise ScrapliTypeError(f"`{caller}` expects a string path to a file, got {type(file)}")
        resolved_file = resolve_file(file)

        with open(resolved_file, "r", encoding="utf-8") as f:
            commands = f.read().splitlines()

        return commands

    @classmethod
    def _pre_send_interactive(
        cls,
        host: str,
        interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]],
        failed_when_contains: Optional[Union[str, List[str]]] = None,
    ) -> Response:
        """
        Handle pre "send_interactive" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            interact_events: list of tuples containing the "interactions" with the device
                each list element must have an input and an expected response, and may have an
                optional bool for the third and final element -- the optional bool specifies if the
                input that is sent to the device is "hidden" (ex: password), if the hidden param is
                not provided it is assumed the input is "normal" (not hidden)
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        joined_input = ", ".join([event[0] for event in interact_events])
        return cls._pre_send_command(
            host=host, command=joined_input, failed_when_contains=failed_when_contains
        )
        
    

Classes

BaseGenericDriver

Expand source code
        
class BaseGenericDriver:
    @staticmethod
    def _pre_send_command(
        host: str, command: str, failed_when_contains: Optional[Union[str, List[str]]] = None
    ) -> Response:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            command: string to send to device in privilege exec mode
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(command, str):
            raise ScrapliTypeError(
                f"`send_command` expects a single string, got {type(command)}, "
                "to send a list of commands use the `send_commands` method instead."
            )

        response = Response(
            host=host,
            channel_input=command,
            failed_when_contains=failed_when_contains,
        )

        return response

    @staticmethod
    def _post_send_command(
        raw_response: bytes, processed_response: bytes, response: Response
    ) -> Response:
        """
        Handle post "send_command" tasks for consistency between sync/async versions

        Args:
            raw_response: raw response returned from the channel
            processed_response: processed response returned from the channel
            response: response object to update with channel results

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        response.record_response(result=processed_response)
        response.raw_result = raw_response
        return response

    @staticmethod
    def _pre_send_commands(commands: List[str]) -> MultiResponse:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            commands: list of strings to send to device in privilege exec mode

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(commands, list):
            raise ScrapliTypeError(
                f"`send_commands` expects a list of strings, got {type(commands)}, "
                "to send a single command use the `send_command` method instead."
            )

        responses = MultiResponse()

        return responses

    @staticmethod
    def _pre_send_from_file(file: str, caller: str) -> List[str]:
        """
        Handle pre "send_*_from_file" tasks for consistency between sync/async versions

        Args:
            file: string path to file
            caller: name of the calling method for more helpful error message

        Returns:
            list: list of commands/configs read from file

        Raises:
            ScrapliTypeError: if anything but a string is provided for `file`

        """
        if not isinstance(file, str):
            raise ScrapliTypeError(f"`{caller}` expects a string path to a file, got {type(file)}")
        resolved_file = resolve_file(file)

        with open(resolved_file, "r", encoding="utf-8") as f:
            commands = f.read().splitlines()

        return commands

    @classmethod
    def _pre_send_interactive(
        cls,
        host: str,
        interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]],
        failed_when_contains: Optional[Union[str, List[str]]] = None,
    ) -> Response:
        """
        Handle pre "send_interactive" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            interact_events: list of tuples containing the "interactions" with the device
                each list element must have an input and an expected response, and may have an
                optional bool for the third and final element -- the optional bool specifies if the
                input that is sent to the device is "hidden" (ex: password), if the hidden param is
                not provided it is assumed the input is "normal" (not hidden)
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        joined_input = ", ".join([event[0] for event in interact_events])
        return cls._pre_send_command(
            host=host, command=joined_input, failed_when_contains=failed_when_contains
        )
        
    

Descendants

  • scrapli.driver.generic.async_driver.AsyncGenericDriver
  • scrapli.driver.generic.sync_driver.GenericDriver