Skip to content

Module scrapli.response

scrapli.response

Expand source code
        
"""scrapli.response"""
from collections import UserList
from datetime import datetime
from io import TextIOWrapper
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union

from scrapli.exceptions import ScrapliCommandFailure
from scrapli.helper import _textfsm_get_template, genie_parse, textfsm_parse, ttp_parse


class Response:
    def __init__(
        self,
        host: str,
        channel_input: str,
        textfsm_platform: str = "",
        genie_platform: str = "",
        failed_when_contains: Optional[Union[str, List[str]]] = None,
    ):
        """
        Scrapli Response

        Store channel_input, resulting output, and start/end/elapsed time information. Attempt to
        determine if command was successful or not and reflect that in a failed attribute.

        Args:
            host: host that was operated on
            channel_input: input that got sent down the channel
            textfsm_platform: ntc-templates friendly platform type
            genie_platform: cisco pyats/genie friendly platform type
            failed_when_contains: list of strings that, if present in final output, represent a
                failed command/interaction

        Returns:
            None

        Raises:
            N/A

        """
        self.host = host
        self.start_time = datetime.now()
        self.finish_time: Optional[datetime] = None
        self.elapsed_time: Optional[float] = None

        self.channel_input = channel_input
        self.textfsm_platform = textfsm_platform
        self.genie_platform = genie_platform
        self.raw_result: bytes = b""
        self.result: str = ""

        if isinstance(failed_when_contains, str):
            failed_when_contains = [failed_when_contains]
        self.failed_when_contains = failed_when_contains
        self.failed = True

    def __bool__(self) -> bool:
        """
        Magic bool method based on channel_input being failed or not

        Args:
            N/A

        Returns:
            bool: True/False if channel_input failed

        Raises:
            N/A

        """
        return self.failed

    def __repr__(self) -> str:
        """
        Magic repr method for Response class

        Args:
            N/A

        Returns:
            str: repr for class object

        Raises:
            N/A

        """
        return (
            f"{self.__class__.__name__}("
            f"host={self.host!r},"
            f"channel_input={self.channel_input!r},"
            f"textfsm_platform={self.textfsm_platform!r},"
            f"genie_platform={self.genie_platform!r},"
            f"failed_when_contains={self.failed_when_contains!r})"
        )

    def __str__(self) -> str:
        """
        Magic str method for Response class

        Args:
            N/A

        Returns:
            str: str for class object

        Raises:
            N/A

        """
        return f"{self.__class__.__name__} "

    def record_response(self, result: bytes) -> None:
        """
        Record channel_input results and elapsed time of channel input/reading output

        Args:
            result: string result of channel_input

        Returns:
            None

        Raises:
            N/A

        """
        self.finish_time = datetime.now()
        self.elapsed_time = (self.finish_time - self.start_time).total_seconds()
        self.raw_result = result

        try:
            self.result = result.decode()
        except UnicodeDecodeError:
            # sometimes we get some "garbage" characters, the iso encoding seems to handle these
            # better but unclear what the other impact is so we'll just catch exceptions and try
            # this encoding
            self.result = result.decode(encoding="ISO-8859-1")

        if not self.failed_when_contains:
            self.failed = False
        elif not any(err in self.result for err in self.failed_when_contains):
            self.failed = False

    def textfsm_parse_output(self, to_dict: bool = True) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with textfsm, always return structured data

        Returns an empty list if parsing fails!

        Args:
            to_dict: convert textfsm output from list of lists to list of dicts -- basically create
                dict from header and row data so it is easier to read/parse the output

        Returns:
            structured_result: empty list or parsed data from textfsm

        Raises:
            N/A

        """
        template = _textfsm_get_template(platform=self.textfsm_platform, command=self.channel_input)
        if isinstance(template, TextIOWrapper):
            structured_result = (
                textfsm_parse(template=template, output=self.result, to_dict=to_dict) or []
            )
        else:
            structured_result = []
        return structured_result

    def genie_parse_output(self) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with genie, always return structured data

        Returns an empty list if parsing fails!

        Args:
            N/A

        Returns:
            structured_result: empty list or parsed data from genie

        Raises:
            N/A

        """
        structured_result = genie_parse(
            platform=self.genie_platform, command=self.channel_input, output=self.result
        )
        return structured_result

    def ttp_parse_output(
        self, template: Union[str, TextIOWrapper]
    ) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with ttp, always return structured data

        Returns an empty list if parsing fails!

        Args:
            template: string path to ttp template or opened ttp template file

        Returns:
            structured_result: empty list or parsed data from ttp

        Raises:
            N/A

        """
        structured_result = ttp_parse(template=template, output=self.result) or []
        return structured_result

    def raise_for_status(self) -> None:
        """
        Raise a `ScrapliCommandFailure` if command/config failed

        Args:
            N/A

        Returns:
            None

        Raises:
            ScrapliCommandFailure: if command/config failed

        """
        if self.failed:
            raise ScrapliCommandFailure()


if TYPE_CHECKING:
    ScrapliMultiResponse = UserList[Response]  # pylint:  disable=E1136; # pragma:  no cover
else:
    ScrapliMultiResponse = UserList


class MultiResponse(ScrapliMultiResponse):
    def __str__(self) -> str:
        """
        Magic str method for MultiResponse class

        Args:
            N/A

        Returns:
            str: str for class object

        Raises:
            N/A

        """
        return (
            f"{self.__class__.__name__} "
        )

    @property
    def host(self) -> str:
        """
        Return the host of the multiresponse

        Args:
            N/A

        Returns:
            str: The host of the associated responses

        Raises:
            N/A

        """
        try:
            response = self.data[0]
        except IndexError:
            return ""
        host = response.host
        return host

    @property
    def failed(self) -> bool:
        """
        Determine if any elements of MultiResponse are failed

        Args:
            N/A

        Returns:
            bool: True for failed

        Raises:
            N/A

        """
        if any(response.failed for response in self.data):
            return True
        return False

    @property
    def result(self) -> str:
        """
        Build a unified result from all elements of MultiResponse

        Args:
            N/A

        Returns:
            str: Unified result by combining results of all elements of MultiResponse

        Raises:
            N/A

        """
        result = ""
        for response in self.data:
            result += "\n".join([response.channel_input, response.result])
        return result

    def raise_for_status(self) -> None:
        """
        Raise a `ScrapliCommandFailure` if any elements are failed

        Args:
            N/A

        Returns:
            None

        Raises:
            ScrapliCommandFailure: if any elements are failed

        """
        if self.failed:
            raise ScrapliCommandFailure()
        
    

Classes

MultiResponse

1
A more or less complete user-defined wrapper around list objects.
Expand source code
        
class MultiResponse(ScrapliMultiResponse):
    def __str__(self) -> str:
        """
        Magic str method for MultiResponse class

        Args:
            N/A

        Returns:
            str: str for class object

        Raises:
            N/A

        """
        return (
            f"{self.__class__.__name__} "
        )

    @property
    def host(self) -> str:
        """
        Return the host of the multiresponse

        Args:
            N/A

        Returns:
            str: The host of the associated responses

        Raises:
            N/A

        """
        try:
            response = self.data[0]
        except IndexError:
            return ""
        host = response.host
        return host

    @property
    def failed(self) -> bool:
        """
        Determine if any elements of MultiResponse are failed

        Args:
            N/A

        Returns:
            bool: True for failed

        Raises:
            N/A

        """
        if any(response.failed for response in self.data):
            return True
        return False

    @property
    def result(self) -> str:
        """
        Build a unified result from all elements of MultiResponse

        Args:
            N/A

        Returns:
            str: Unified result by combining results of all elements of MultiResponse

        Raises:
            N/A

        """
        result = ""
        for response in self.data:
            result += "\n".join([response.channel_input, response.result])
        return result

    def raise_for_status(self) -> None:
        """
        Raise a `ScrapliCommandFailure` if any elements are failed

        Args:
            N/A

        Returns:
            None

        Raises:
            ScrapliCommandFailure: if any elements are failed

        """
        if self.failed:
            raise ScrapliCommandFailure()
        
    

Ancestors (in MRO)

  • collections.UserList
  • collections.abc.MutableSequence
  • collections.abc.Sequence
  • collections.abc.Reversible
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

failed: bool

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Determine if any elements of MultiResponse are failed

Args:
    N/A

Returns:
    bool: True for failed

Raises:
    N/A

host: str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Return the host of the multiresponse

Args:
    N/A

Returns:
    str: The host of the associated responses

Raises:
    N/A

result: str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Build a unified result from all elements of MultiResponse

Args:
    N/A

Returns:
    str: Unified result by combining results of all elements of MultiResponse

Raises:
    N/A

Methods

raise_for_status

raise_for_status(self) ‑> None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Raise a `ScrapliCommandFailure` if any elements are failed

Args:
    N/A

Returns:
    None

Raises:
    ScrapliCommandFailure: if any elements are failed

Response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
Scrapli Response

Store channel_input, resulting output, and start/end/elapsed time information. Attempt to
determine if command was successful or not and reflect that in a failed attribute.

Args:
    host: host that was operated on
    channel_input: input that got sent down the channel
    textfsm_platform: ntc-templates friendly platform type
    genie_platform: cisco pyats/genie friendly platform type
    failed_when_contains: list of strings that, if present in final output, represent a
        failed command/interaction

Returns:
    None

Raises:
    N/A
Expand source code
        
class Response:
    def __init__(
        self,
        host: str,
        channel_input: str,
        textfsm_platform: str = "",
        genie_platform: str = "",
        failed_when_contains: Optional[Union[str, List[str]]] = None,
    ):
        """
        Scrapli Response

        Store channel_input, resulting output, and start/end/elapsed time information. Attempt to
        determine if command was successful or not and reflect that in a failed attribute.

        Args:
            host: host that was operated on
            channel_input: input that got sent down the channel
            textfsm_platform: ntc-templates friendly platform type
            genie_platform: cisco pyats/genie friendly platform type
            failed_when_contains: list of strings that, if present in final output, represent a
                failed command/interaction

        Returns:
            None

        Raises:
            N/A

        """
        self.host = host
        self.start_time = datetime.now()
        self.finish_time: Optional[datetime] = None
        self.elapsed_time: Optional[float] = None

        self.channel_input = channel_input
        self.textfsm_platform = textfsm_platform
        self.genie_platform = genie_platform
        self.raw_result: bytes = b""
        self.result: str = ""

        if isinstance(failed_when_contains, str):
            failed_when_contains = [failed_when_contains]
        self.failed_when_contains = failed_when_contains
        self.failed = True

    def __bool__(self) -> bool:
        """
        Magic bool method based on channel_input being failed or not

        Args:
            N/A

        Returns:
            bool: True/False if channel_input failed

        Raises:
            N/A

        """
        return self.failed

    def __repr__(self) -> str:
        """
        Magic repr method for Response class

        Args:
            N/A

        Returns:
            str: repr for class object

        Raises:
            N/A

        """
        return (
            f"{self.__class__.__name__}("
            f"host={self.host!r},"
            f"channel_input={self.channel_input!r},"
            f"textfsm_platform={self.textfsm_platform!r},"
            f"genie_platform={self.genie_platform!r},"
            f"failed_when_contains={self.failed_when_contains!r})"
        )

    def __str__(self) -> str:
        """
        Magic str method for Response class

        Args:
            N/A

        Returns:
            str: str for class object

        Raises:
            N/A

        """
        return f"{self.__class__.__name__} "

    def record_response(self, result: bytes) -> None:
        """
        Record channel_input results and elapsed time of channel input/reading output

        Args:
            result: string result of channel_input

        Returns:
            None

        Raises:
            N/A

        """
        self.finish_time = datetime.now()
        self.elapsed_time = (self.finish_time - self.start_time).total_seconds()
        self.raw_result = result

        try:
            self.result = result.decode()
        except UnicodeDecodeError:
            # sometimes we get some "garbage" characters, the iso encoding seems to handle these
            # better but unclear what the other impact is so we'll just catch exceptions and try
            # this encoding
            self.result = result.decode(encoding="ISO-8859-1")

        if not self.failed_when_contains:
            self.failed = False
        elif not any(err in self.result for err in self.failed_when_contains):
            self.failed = False

    def textfsm_parse_output(self, to_dict: bool = True) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with textfsm, always return structured data

        Returns an empty list if parsing fails!

        Args:
            to_dict: convert textfsm output from list of lists to list of dicts -- basically create
                dict from header and row data so it is easier to read/parse the output

        Returns:
            structured_result: empty list or parsed data from textfsm

        Raises:
            N/A

        """
        template = _textfsm_get_template(platform=self.textfsm_platform, command=self.channel_input)
        if isinstance(template, TextIOWrapper):
            structured_result = (
                textfsm_parse(template=template, output=self.result, to_dict=to_dict) or []
            )
        else:
            structured_result = []
        return structured_result

    def genie_parse_output(self) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with genie, always return structured data

        Returns an empty list if parsing fails!

        Args:
            N/A

        Returns:
            structured_result: empty list or parsed data from genie

        Raises:
            N/A

        """
        structured_result = genie_parse(
            platform=self.genie_platform, command=self.channel_input, output=self.result
        )
        return structured_result

    def ttp_parse_output(
        self, template: Union[str, TextIOWrapper]
    ) -> Union[Dict[str, Any], List[Any]]:
        """
        Parse results with ttp, always return structured data

        Returns an empty list if parsing fails!

        Args:
            template: string path to ttp template or opened ttp template file

        Returns:
            structured_result: empty list or parsed data from ttp

        Raises:
            N/A

        """
        structured_result = ttp_parse(template=template, output=self.result) or []
        return structured_result

    def raise_for_status(self) -> None:
        """
        Raise a `ScrapliCommandFailure` if command/config failed

        Args:
            N/A

        Returns:
            None

        Raises:
            ScrapliCommandFailure: if command/config failed

        """
        if self.failed:
            raise ScrapliCommandFailure()
        
    

Methods

genie_parse_output

genie_parse_output(self) ‑> Union[List[Any], Dict[str, Any]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
Parse results with genie, always return structured data

Returns an empty list if parsing fails!

Args:
    N/A

Returns:
    structured_result: empty list or parsed data from genie

Raises:
    N/A
raise_for_status

raise_for_status(self) ‑> None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Raise a `ScrapliCommandFailure` if command/config failed

Args:
    N/A

Returns:
    None

Raises:
    ScrapliCommandFailure: if command/config failed
record_response

record_response(self, result: bytes) ‑> None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Record channel_input results and elapsed time of channel input/reading output

Args:
    result: string result of channel_input

Returns:
    None

Raises:
    N/A
textfsm_parse_output

textfsm_parse_output(self, to_dict: bool = True) ‑> Union[List[Any], Dict[str, Any]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
Parse results with textfsm, always return structured data

Returns an empty list if parsing fails!

Args:
    to_dict: convert textfsm output from list of lists to list of dicts -- basically create
        dict from header and row data so it is easier to read/parse the output

Returns:
    structured_result: empty list or parsed data from textfsm

Raises:
    N/A
ttp_parse_output

ttp_parse_output(self, template: Union[str, _io.TextIOWrapper]) ‑> Union[List[Any], Dict[str, Any]]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
Parse results with ttp, always return structured data

Returns an empty list if parsing fails!

Args:
    template: string path to ttp template or opened ttp template file

Returns:
    structured_result: empty list or parsed data from ttp

Raises:
    N/A