factory

scrapli.factory

AsyncScrapli

Bases: AsyncNetworkDriver

Source code in scrapli/factory.py

class AsyncScrapli(AsyncNetworkDriver):
    CORE_PLATFORM_MAP = {
        "arista_eos": AsyncEOSDriver,
        "cisco_iosxe": AsyncIOSXEDriver,
        "cisco_iosxr": AsyncIOSXRDriver,
        "cisco_nxos": AsyncNXOSDriver,
        "juniper_junos": AsyncJunosDriver,
    }
    DRIVER_MAP = {"network": AsyncNetworkDriver, "generic": AsyncGenericDriver}

    @classmethod
    def _get_driver_class(
        cls, platform_details: Dict[str, Any], variant: Optional[str]
    ) -> Union[Type[AsyncNetworkDriver], Type[AsyncGenericDriver]]:
        """
        Fetch community driver class based on platform details

        Args:
            platform_details: dict of details about community platform from scrapli_community
                library
            variant: optional name of variant of community platform

        Returns:
            NetworkDriver: final driver class

        Raises:
            N/A

        """
        final_driver: Union[
            Type[AsyncNetworkDriver],
            Type[AsyncGenericDriver],
        ]

        if variant and platform_details["variants"][variant].get("driver_type"):
            variant_driver_data = platform_details["variants"][variant].pop("driver_type")
            final_driver = variant_driver_data["async"]
            return final_driver

        if isinstance(platform_details["driver_type"], str):
            driver_type = platform_details["driver_type"]
            standard_final_driver = cls.DRIVER_MAP.get(driver_type, None)
            if standard_final_driver:
                return standard_final_driver

        final_driver = platform_details["driver_type"]["async"]
        return final_driver

    @classmethod
    def _get_community_driver(
        cls, community_platform_name: str, variant: Optional[str]
    ) -> Tuple[Union[Type[AsyncNetworkDriver], Type[AsyncGenericDriver]], Dict[str, Any]]:
        """
        Get community driver

        Args:
            community_platform_name: name of community
            variant: optional name of variant of community platform

        Returns:
            NetworkDriver: final driver class

        Raises:
            N/A

        """
        platform_details = _get_community_platform_details(
            community_platform_name=community_platform_name
        )

        final_driver = cls._get_driver_class(platform_details=platform_details, variant=variant)
        final_platform_kwargs = _get_driver_kwargs(
            platform_details=platform_details, variant=variant, _async=True
        )

        return final_driver, final_platform_kwargs

    @classmethod
    def _get_driver(
        cls, platform: str, variant: Optional[str]
    ) -> Tuple[Union[Type[AsyncNetworkDriver], Type[AsyncGenericDriver]], Dict[str, Any]]:
        """
        Parent get driver method for sync Scrapli

        Args:
            platform: name of target platform; i.e. `cisco_iosxe`, `arista_eos`, etc.
            variant: name of the target platform variant

        Returns:
            NetworkDriver: final driver class; generally NetworkDriver, but for some community
                platforms could be GenericDriver, also returns any additional kwargs comming from
                the community platform (if any)

        Raises:
            N/A

        """
        additional_kwargs: Dict[str, Any] = {}
        final_driver: Union[Type[AsyncGenericDriver], Type[AsyncNetworkDriver]]

        if platform in cls.CORE_PLATFORM_MAP:
            final_driver = cls.CORE_PLATFORM_MAP[platform]
            msg = f"Driver '{final_driver}' selected from scrapli core drivers"
        else:
            final_driver, additional_kwargs = cls._get_community_driver(
                community_platform_name=platform, variant=variant
            )
            msg = (
                f"Driver '{final_driver}' selected from scrapli community platforms, with the "
                f"following platform arguments: '{additional_kwargs}'"
            )

        logger.info(msg)
        return final_driver, additional_kwargs

    def __new__(  # pylint: disable=R0914
        cls,
        platform: str,
        host: str,
        privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
        default_desired_privilege_level: Optional[str] = None,
        port: Optional[int] = None,
        auth_username: Optional[str] = None,
        auth_password: Optional[str] = None,
        auth_private_key: Optional[str] = None,
        auth_private_key_passphrase: Optional[str] = None,
        auth_strict_key: Optional[bool] = None,
        auth_bypass: Optional[bool] = None,
        timeout_socket: Optional[float] = None,
        timeout_transport: Optional[float] = None,
        timeout_ops: Optional[float] = None,
        comms_return_char: Optional[str] = None,
        ssh_config_file: Optional[Union[str, bool]] = None,
        ssh_known_hosts_file: Optional[Union[str, bool]] = None,
        on_init: Optional[Callable[..., Any]] = None,
        on_open: Optional[Callable[..., Any]] = None,
        on_close: Optional[Callable[..., Any]] = None,
        transport: Optional[str] = None,
        transport_options: Optional[Dict[str, Any]] = None,
        channel_log: Optional[Union[str, bool, BytesIO]] = None,
        channel_log_mode: Optional[str] = None,
        channel_lock: Optional[bool] = None,
        logging_uid: Optional[str] = None,
        auth_secondary: Optional[str] = None,
        failed_when_contains: Optional[List[str]] = None,
        textfsm_platform: Optional[str] = None,
        genie_platform: Optional[str] = None,
        variant: Optional[str] = None,
        **kwargs: Dict[Any, Any],
    ) -> "AsyncScrapli":
        r"""
        Scrapli Factory method for asynchronous drivers

        Args:
            platform: name of the scrapli platform to return a connection object for; should be
                one of the "core" platforms or a valid community platform name
            host: host ip/name to connect to
            port: port to connect to
            auth_username: username for authentication
            auth_private_key: path to private key for authentication
            auth_private_key_passphrase: passphrase for decrypting ssh key if necessary
            auth_password: password for authentication
            auth_strict_key: strict host checking or not
            auth_bypass: bypass "in channel" authentication -- only supported with telnet,
                asynctelnet, and system transport plugins
            timeout_socket: timeout for establishing socket/initial connection in seconds
            timeout_transport: timeout for ssh|telnet transport in seconds
            timeout_ops: timeout for ssh channel operations
            comms_return_char: character to use to send returns to host
            ssh_config_file: string to path for ssh config file, True to use default ssh config file
                or False to ignore default ssh config file
            ssh_known_hosts_file: string to path for ssh known hosts file, True to use default known
                file locations. Only applicable/needed if `auth_strict_key` is set to True
            on_init: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed as the last step of object instantiation -- its purpose is
                primarily to provide a mechanism for scrapli community platforms to have an easy way
                to modify initialization arguments/object attributes without needing to create a
                class that extends the driver, instead allowing the community platforms to simply
                build from the GenericDriver or NetworkDriver classes, and pass this callable to do
                things such as appending to a username (looking at you RouterOS!!). Note that this
                is *always* a synchronous function (even for asyncio drivers)!
            on_open: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately after authentication is completed. Common use
                cases for this callable would be to disable paging or accept any kind of banner
                message that prompts a user upon connection
            on_close: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately prior to closing the underlying transport.
                Common use cases for this callable would be to save configurations prior to exiting,
                or to logout properly to free up vtys or similar
            transport: name of the transport plugin to use for the actual telnet/ssh/netconf
                connection. Available "core" transports are:
                    - system
                    - telnet
                    - asynctelnet
                    - ssh2
                    - paramiko
                    - asyncssh
                Please see relevant transport plugin section for details. Additionally third party
                transport plugins may be available.
            transport_options: dictionary of options to pass to selected transport class; see
                docs for given transport class for details of what to pass here
            channel_lock: True/False to lock the channel (threading.Lock/asyncio.Lock) during
                any channel operations, defaults to False
            channel_log: True/False or a string path to a file of where to write out channel logs --
                these are not "logs" in the normal logging module sense, but only the output that is
                read from the channel. In other words, the output of the channel log should look
                similar to what you would see as a human connecting to a device
            channel_log_mode: "write"|"append", all other values will raise ValueError,
                does what it sounds like it should by setting the channel log to the provided mode
            logging_uid: unique identifier (string) to associate to log messages; useful if you have
                multiple connections to the same device (i.e. one console, one ssh, or one to each
                supervisor module, etc.)
            failed_when_contains: list of strings indicating command/config failure
            textfsm_platform: string to use to fetch ntc-templates templates for textfsm parsing
            genie_platform: string to use to fetch genie parser templates
            privilege_levels: optional user provided privilege levels, if left None will default to
                scrapli standard privilege levels
            default_desired_privilege_level: string of name of default desired priv, this is the
                priv level that is generally used to disable paging/set terminal width and things
                like that upon first login, and is also the priv level scrapli will try to acquire
                for normal "command" operations (`send_command`, `send_commands`)
            auth_secondary: password to use for secondary authentication (enable)
            failed_when_contains: List of strings that indicate a command/config has failed
            variant: name of the community platform variant if desired
            **kwargs: should be unused, but here to accept any additional kwargs from users

        Returns:
            final_driver: asynchronous driver class for provided driver

        Raises:
            ScrapliValueError: if provided transport is asyncio
            ScrapliTypeError: if `platform` not in keyword arguments

        """
        logger.debug("AsyncScrapli factory initialized")

        if transport not in ASYNCIO_TRANSPORTS:
            raise ScrapliValueError("Use 'Scrapli' if using a synchronous transport!")

        if not isinstance(platform, str):
            raise ScrapliTypeError(f"Argument 'platform' must be 'str' got '{type(platform)}'")

        provided_kwargs = _build_provided_kwargs_dict(
            host=host,
            port=port,
            auth_username=auth_username,
            auth_password=auth_password,
            auth_private_key=auth_private_key,
            auth_private_key_passphrase=auth_private_key_passphrase,
            auth_strict_key=auth_strict_key,
            auth_bypass=auth_bypass,
            timeout_socket=timeout_socket,
            timeout_transport=timeout_transport,
            timeout_ops=timeout_ops,
            comms_return_char=comms_return_char,
            ssh_config_file=ssh_config_file,
            ssh_known_hosts_file=ssh_known_hosts_file,
            on_init=on_init,
            on_open=on_open,
            on_close=on_close,
            transport=transport,
            transport_options=transport_options,
            channel_log=channel_log,
            channel_log_mode=channel_log_mode,
            channel_lock=channel_lock,
            logging_uid=logging_uid,
            privilege_levels=privilege_levels,
            default_desired_privilege_level=default_desired_privilege_level,
            auth_secondary=auth_secondary,
            failed_when_contains=failed_when_contains,
            textfsm_platform=textfsm_platform,
            genie_platform=genie_platform,
            **kwargs,
        )

        final_driver, additional_kwargs = cls._get_driver(platform=platform, variant=variant)

        # at this point will need to merge the additional kwargs in (for community drivers),
        # ensure that kwargs passed by user supersede the ones coming from community platform
        if additional_kwargs:
            final_kwargs = {**additional_kwargs, **provided_kwargs}
        else:
            final_kwargs = provided_kwargs

        final_conn = final_driver(**final_kwargs)
        # cast the final conn to type Scrapli to appease mypy -- we know it will be a NetworkDriver
        # or GenericDriver, but thats ok =)
        final_conn = cast(AsyncScrapli, final_conn)
        return final_conn

__new__(platform: str, host: str, privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None, default_desired_privilege_level: Optional[str] = None, port: Optional[int] = None, auth_username: Optional[str] = None, auth_password: Optional[str] = None, auth_private_key: Optional[str] = None, auth_private_key_passphrase: Optional[str] = None, auth_strict_key: Optional[bool] = None, auth_bypass: Optional[bool] = None, timeout_socket: Optional[float] = None, timeout_transport: Optional[float] = None, timeout_ops: Optional[float] = None, comms_return_char: Optional[str] = None, ssh_config_file: Optional[Union[str, bool]] = None, ssh_known_hosts_file: Optional[Union[str, bool]] = None, on_init: Optional[Callable[..., Any]] = None, on_open: Optional[Callable[..., Any]] = None, on_close: Optional[Callable[..., Any]] = None, transport: Optional[str] = None, transport_options: Optional[Dict[str, Any]] = None, channel_log: Optional[Union[str, bool, BytesIO]] = None, channel_log_mode: Optional[str] = None, channel_lock: Optional[bool] = None, logging_uid: Optional[str] = None, auth_secondary: Optional[str] = None, failed_when_contains: Optional[List[str]] = None, textfsm_platform: Optional[str] = None, genie_platform: Optional[str] = None, variant: Optional[str] = None, **kwargs: Dict[Any, Any]) -> AsyncScrapli

Scrapli Factory method for asynchronous drivers

Parameters:

Name	Type	Description	Default
platform	str	name of the scrapli platform to return a connection object for; should be one of the "core" platforms or a valid community platform name	required
host	str	host ip/name to connect to	required
port	Optional[int]	port to connect to	None
auth_username	Optional[str]	username for authentication	None
auth_private_key	Optional[str]	path to private key for authentication	None
auth_private_key_passphrase	Optional[str]	passphrase for decrypting ssh key if necessary	None
auth_password	Optional[str]	password for authentication	None
auth_strict_key	Optional[bool]	strict host checking or not	None
auth_bypass	Optional[bool]	bypass "in channel" authentication -- only supported with telnet, asynctelnet, and system transport plugins	None
timeout_socket	Optional[float]	timeout for establishing socket/initial connection in seconds	None
timeout_transport	Optional[float]	timeout for ssh|telnet transport in seconds	None
timeout_ops	Optional[float]	timeout for ssh channel operations	None
comms_return_char	Optional[str]	character to use to send returns to host	None
ssh_config_file	Optional[Union[str, bool]]	string to path for ssh config file, True to use default ssh config file or False to ignore default ssh config file	None
ssh_known_hosts_file	Optional[Union[str, bool]]	string to path for ssh known hosts file, True to use default known file locations. Only applicable/needed if auth_strict_key is set to True	None
on_init	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed as the last step of object instantiation -- its purpose is primarily to provide a mechanism for scrapli community platforms to have an easy way to modify initialization arguments/object attributes without needing to create a class that extends the driver, instead allowing the community platforms to simply build from the GenericDriver or NetworkDriver classes, and pass this callable to do things such as appending to a username (looking at you RouterOS!!). Note that this is always a synchronous function (even for asyncio drivers)!	None
on_open	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed immediately after authentication is completed. Common use cases for this callable would be to disable paging or accept any kind of banner message that prompts a user upon connection	None
on_close	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed immediately prior to closing the underlying transport. Common use cases for this callable would be to save configurations prior to exiting, or to logout properly to free up vtys or similar	None
transport	Optional[str]	name of the transport plugin to use for the actual telnet/ssh/netconf connection. Available "core" transports are: - system - telnet - asynctelnet - ssh2 - paramiko - asyncssh Please see relevant transport plugin section for details. Additionally third party transport plugins may be available.	None
transport_options	Optional[Dict[str, Any]]	dictionary of options to pass to selected transport class; see docs for given transport class for details of what to pass here	None
channel_lock	Optional[bool]	True/False to lock the channel (threading.Lock/asyncio.Lock) during any channel operations, defaults to False	None
channel_log	Optional[Union[str, bool, BytesIO]]	True/False or a string path to a file of where to write out channel logs -- these are not "logs" in the normal logging module sense, but only the output that is read from the channel. In other words, the output of the channel log should look similar to what you would see as a human connecting to a device	None
channel_log_mode	Optional[str]	"write"|"append", all other values will raise ValueError, does what it sounds like it should by setting the channel log to the provided mode	None
logging_uid	Optional[str]	unique identifier (string) to associate to log messages; useful if you have multiple connections to the same device (i.e. one console, one ssh, or one to each supervisor module, etc.)	None
failed_when_contains	Optional[List[str]]	list of strings indicating command/config failure	None
textfsm_platform	Optional[str]	string to use to fetch ntc-templates templates for textfsm parsing	None
genie_platform	Optional[str]	string to use to fetch genie parser templates	None
privilege_levels	Optional[Dict[str, PrivilegeLevel]]	optional user provided privilege levels, if left None will default to scrapli standard privilege levels	None
default_desired_privilege_level	Optional[str]	string of name of default desired priv, this is the priv level that is generally used to disable paging/set terminal width and things like that upon first login, and is also the priv level scrapli will try to acquire for normal "command" operations (send_command, send_commands)	None
auth_secondary	Optional[str]	password to use for secondary authentication (enable)	None
failed_when_contains	Optional[List[str]]	List of strings that indicate a command/config has failed	None
variant	Optional[str]	name of the community platform variant if desired	None
**kwargs	Dict[Any, Any]	should be unused, but here to accept any additional kwargs from users	{}

Returns:

Name	Type	Description
final_driver	AsyncScrapli	asynchronous driver class for provided driver

Raises:

Type	Description
ScrapliValueError	if provided transport is asyncio
ScrapliTypeError	if platform not in keyword arguments

Source code in scrapli/factory.py

def __new__(  # pylint: disable=R0914
    cls,
    platform: str,
    host: str,
    privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
    default_desired_privilege_level: Optional[str] = None,
    port: Optional[int] = None,
    auth_username: Optional[str] = None,
    auth_password: Optional[str] = None,
    auth_private_key: Optional[str] = None,
    auth_private_key_passphrase: Optional[str] = None,
    auth_strict_key: Optional[bool] = None,
    auth_bypass: Optional[bool] = None,
    timeout_socket: Optional[float] = None,
    timeout_transport: Optional[float] = None,
    timeout_ops: Optional[float] = None,
    comms_return_char: Optional[str] = None,
    ssh_config_file: Optional[Union[str, bool]] = None,
    ssh_known_hosts_file: Optional[Union[str, bool]] = None,
    on_init: Optional[Callable[..., Any]] = None,
    on_open: Optional[Callable[..., Any]] = None,
    on_close: Optional[Callable[..., Any]] = None,
    transport: Optional[str] = None,
    transport_options: Optional[Dict[str, Any]] = None,
    channel_log: Optional[Union[str, bool, BytesIO]] = None,
    channel_log_mode: Optional[str] = None,
    channel_lock: Optional[bool] = None,
    logging_uid: Optional[str] = None,
    auth_secondary: Optional[str] = None,
    failed_when_contains: Optional[List[str]] = None,
    textfsm_platform: Optional[str] = None,
    genie_platform: Optional[str] = None,
    variant: Optional[str] = None,
    **kwargs: Dict[Any, Any],
) -> "AsyncScrapli":
    r"""
    Scrapli Factory method for asynchronous drivers

    Args:
        platform: name of the scrapli platform to return a connection object for; should be
            one of the "core" platforms or a valid community platform name
        host: host ip/name to connect to
        port: port to connect to
        auth_username: username for authentication
        auth_private_key: path to private key for authentication
        auth_private_key_passphrase: passphrase for decrypting ssh key if necessary
        auth_password: password for authentication
        auth_strict_key: strict host checking or not
        auth_bypass: bypass "in channel" authentication -- only supported with telnet,
            asynctelnet, and system transport plugins
        timeout_socket: timeout for establishing socket/initial connection in seconds
        timeout_transport: timeout for ssh|telnet transport in seconds
        timeout_ops: timeout for ssh channel operations
        comms_return_char: character to use to send returns to host
        ssh_config_file: string to path for ssh config file, True to use default ssh config file
            or False to ignore default ssh config file
        ssh_known_hosts_file: string to path for ssh known hosts file, True to use default known
            file locations. Only applicable/needed if `auth_strict_key` is set to True
        on_init: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed as the last step of object instantiation -- its purpose is
            primarily to provide a mechanism for scrapli community platforms to have an easy way
            to modify initialization arguments/object attributes without needing to create a
            class that extends the driver, instead allowing the community platforms to simply
            build from the GenericDriver or NetworkDriver classes, and pass this callable to do
            things such as appending to a username (looking at you RouterOS!!). Note that this
            is *always* a synchronous function (even for asyncio drivers)!
        on_open: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately after authentication is completed. Common use
            cases for this callable would be to disable paging or accept any kind of banner
            message that prompts a user upon connection
        on_close: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately prior to closing the underlying transport.
            Common use cases for this callable would be to save configurations prior to exiting,
            or to logout properly to free up vtys or similar
        transport: name of the transport plugin to use for the actual telnet/ssh/netconf
            connection. Available "core" transports are:
                - system
                - telnet
                - asynctelnet
                - ssh2
                - paramiko
                - asyncssh
            Please see relevant transport plugin section for details. Additionally third party
            transport plugins may be available.
        transport_options: dictionary of options to pass to selected transport class; see
            docs for given transport class for details of what to pass here
        channel_lock: True/False to lock the channel (threading.Lock/asyncio.Lock) during
            any channel operations, defaults to False
        channel_log: True/False or a string path to a file of where to write out channel logs --
            these are not "logs" in the normal logging module sense, but only the output that is
            read from the channel. In other words, the output of the channel log should look
            similar to what you would see as a human connecting to a device
        channel_log_mode: "write"|"append", all other values will raise ValueError,
            does what it sounds like it should by setting the channel log to the provided mode
        logging_uid: unique identifier (string) to associate to log messages; useful if you have
            multiple connections to the same device (i.e. one console, one ssh, or one to each
            supervisor module, etc.)
        failed_when_contains: list of strings indicating command/config failure
        textfsm_platform: string to use to fetch ntc-templates templates for textfsm parsing
        genie_platform: string to use to fetch genie parser templates
        privilege_levels: optional user provided privilege levels, if left None will default to
            scrapli standard privilege levels
        default_desired_privilege_level: string of name of default desired priv, this is the
            priv level that is generally used to disable paging/set terminal width and things
            like that upon first login, and is also the priv level scrapli will try to acquire
            for normal "command" operations (`send_command`, `send_commands`)
        auth_secondary: password to use for secondary authentication (enable)
        failed_when_contains: List of strings that indicate a command/config has failed
        variant: name of the community platform variant if desired
        **kwargs: should be unused, but here to accept any additional kwargs from users

    Returns:
        final_driver: asynchronous driver class for provided driver

    Raises:
        ScrapliValueError: if provided transport is asyncio
        ScrapliTypeError: if `platform` not in keyword arguments

    """
    logger.debug("AsyncScrapli factory initialized")

    if transport not in ASYNCIO_TRANSPORTS:
        raise ScrapliValueError("Use 'Scrapli' if using a synchronous transport!")

    if not isinstance(platform, str):
        raise ScrapliTypeError(f"Argument 'platform' must be 'str' got '{type(platform)}'")

    provided_kwargs = _build_provided_kwargs_dict(
        host=host,
        port=port,
        auth_username=auth_username,
        auth_password=auth_password,
        auth_private_key=auth_private_key,
        auth_private_key_passphrase=auth_private_key_passphrase,
        auth_strict_key=auth_strict_key,
        auth_bypass=auth_bypass,
        timeout_socket=timeout_socket,
        timeout_transport=timeout_transport,
        timeout_ops=timeout_ops,
        comms_return_char=comms_return_char,
        ssh_config_file=ssh_config_file,
        ssh_known_hosts_file=ssh_known_hosts_file,
        on_init=on_init,
        on_open=on_open,
        on_close=on_close,
        transport=transport,
        transport_options=transport_options,
        channel_log=channel_log,
        channel_log_mode=channel_log_mode,
        channel_lock=channel_lock,
        logging_uid=logging_uid,
        privilege_levels=privilege_levels,
        default_desired_privilege_level=default_desired_privilege_level,
        auth_secondary=auth_secondary,
        failed_when_contains=failed_when_contains,
        textfsm_platform=textfsm_platform,
        genie_platform=genie_platform,
        **kwargs,
    )

    final_driver, additional_kwargs = cls._get_driver(platform=platform, variant=variant)

    # at this point will need to merge the additional kwargs in (for community drivers),
    # ensure that kwargs passed by user supersede the ones coming from community platform
    if additional_kwargs:
        final_kwargs = {**additional_kwargs, **provided_kwargs}
    else:
        final_kwargs = provided_kwargs

    final_conn = final_driver(**final_kwargs)
    # cast the final conn to type Scrapli to appease mypy -- we know it will be a NetworkDriver
    # or GenericDriver, but thats ok =)
    final_conn = cast(AsyncScrapli, final_conn)
    return final_conn

Scrapli

Bases: NetworkDriver

Source code in scrapli/factory.py

class Scrapli(NetworkDriver):
    CORE_PLATFORM_MAP = {
        "arista_eos": EOSDriver,
        "cisco_iosxe": IOSXEDriver,
        "cisco_iosxr": IOSXRDriver,
        "cisco_nxos": NXOSDriver,
        "juniper_junos": JunosDriver,
    }
    DRIVER_MAP = {"network": NetworkDriver, "generic": GenericDriver}

    @classmethod
    def _get_driver_class(
        cls, platform_details: Dict[str, Any], variant: Optional[str]
    ) -> Union[Type[NetworkDriver], Type[GenericDriver]]:
        """
        Fetch community driver class based on platform details

        Args:
            platform_details: dict of details about community platform from scrapli_community
                library
            variant: optional name of variant of community platform

        Returns:
            NetworkDriver: final driver class

        Raises:
            N/A

        """
        final_driver: Union[
            Type[NetworkDriver],
            Type[GenericDriver],
        ]

        if variant and platform_details["variants"][variant].get("driver_type"):
            variant_driver_data = platform_details["variants"][variant].pop("driver_type")
            final_driver = variant_driver_data["sync"]
            return final_driver

        if isinstance(platform_details["driver_type"], str):
            driver_type = platform_details["driver_type"]
            standard_final_driver = cls.DRIVER_MAP.get(driver_type, None)
            if standard_final_driver:
                return standard_final_driver

        final_driver = platform_details["driver_type"]["sync"]
        return final_driver

    @classmethod
    def _get_community_driver(
        cls, community_platform_name: str, variant: Optional[str]
    ) -> Tuple[Union[Type[NetworkDriver], Type[GenericDriver]], Dict[str, Any]]:
        """
        Get community driver

        Args:
            community_platform_name: name of community
            variant: optional name of variant of community platform

        Returns:
            NetworkDriver: final driver class

        Raises:
            N/A

        """
        platform_details = _get_community_platform_details(
            community_platform_name=community_platform_name
        )

        final_driver = cls._get_driver_class(platform_details=platform_details, variant=variant)
        final_platform_kwargs = _get_driver_kwargs(
            platform_details=platform_details, variant=variant, _async=False
        )

        return final_driver, final_platform_kwargs

    @classmethod
    def _get_driver(
        cls, platform: str, variant: Optional[str]
    ) -> Tuple[Union[Type[NetworkDriver], Type[GenericDriver]], Dict[str, Any]]:
        """
        Parent get driver method for sync Scrapli

        Args:
            platform: name of target platform; i.e. `cisco_iosxe`, `arista_eos`, etc.
            variant: name of the target platform variant

        Returns:
            NetworkDriver: final driver class; generally NetworkDriver, but for some community
                platforms could be GenericDriver, also returns any additional kwargs comming from
                the community platform (if any)

        Raises:
            N/A

        """
        additional_kwargs: Dict[str, Any] = {}
        final_driver: Union[Type[GenericDriver], Type[NetworkDriver]]

        if platform in cls.CORE_PLATFORM_MAP:
            final_driver = cls.CORE_PLATFORM_MAP[platform]
            msg = f"Driver '{final_driver}' selected from scrapli core drivers"
        else:
            final_driver, additional_kwargs = cls._get_community_driver(
                community_platform_name=platform, variant=variant
            )
            msg = (
                f"Driver '{final_driver}' selected from scrapli community platforms, with the "
                f"following platform arguments: '{additional_kwargs}'"
            )

        logger.info(msg)
        return final_driver, additional_kwargs

    def __new__(  # pylint: disable=R0914
        cls,
        platform: str,
        host: str,
        privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
        default_desired_privilege_level: Optional[str] = None,
        port: Optional[int] = None,
        auth_username: Optional[str] = None,
        auth_password: Optional[str] = None,
        auth_private_key: Optional[str] = None,
        auth_private_key_passphrase: Optional[str] = None,
        auth_strict_key: Optional[bool] = None,
        auth_bypass: Optional[bool] = None,
        timeout_socket: Optional[float] = None,
        timeout_transport: Optional[float] = None,
        timeout_ops: Optional[float] = None,
        comms_return_char: Optional[str] = None,
        ssh_config_file: Optional[Union[str, bool]] = None,
        ssh_known_hosts_file: Optional[Union[str, bool]] = None,
        on_init: Optional[Callable[..., Any]] = None,
        on_open: Optional[Callable[..., Any]] = None,
        on_close: Optional[Callable[..., Any]] = None,
        transport: Optional[str] = None,
        transport_options: Optional[Dict[str, Any]] = None,
        channel_log: Optional[Union[str, bool, BytesIO]] = None,
        channel_lock: Optional[bool] = None,
        channel_log_mode: Optional[str] = None,
        logging_uid: Optional[str] = None,
        auth_secondary: Optional[str] = None,
        failed_when_contains: Optional[List[str]] = None,
        textfsm_platform: Optional[str] = None,
        genie_platform: Optional[str] = None,
        variant: Optional[str] = None,
        **kwargs: Dict[Any, Any],
    ) -> "Scrapli":
        r"""
        Scrapli Factory method for synchronous drivers

        Args:
            platform: name of the scrapli platform to return a connection object for; should be
                one of the "core" platforms or a valid community platform name
            host: host ip/name to connect to
            port: port to connect to
            auth_username: username for authentication
            auth_private_key: path to private key for authentication
            auth_private_key_passphrase: passphrase for decrypting ssh key if necessary
            auth_password: password for authentication
            auth_strict_key: strict host checking or not
            auth_bypass: bypass "in channel" authentication -- only supported with telnet,
                asynctelnet, and system transport plugins
            timeout_socket: timeout for establishing socket/initial connection in seconds
            timeout_transport: timeout for ssh|telnet transport in seconds
            timeout_ops: timeout for ssh channel operations
            comms_return_char: character to use to send returns to host
            ssh_config_file: string to path for ssh config file, True to use default ssh config file
                or False to ignore default ssh config file
            ssh_known_hosts_file: string to path for ssh known hosts file, True to use default known
                file locations. Only applicable/needed if `auth_strict_key` is set to True
            on_init: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed as the last step of object instantiation -- its purpose is
                primarily to provide a mechanism for scrapli community platforms to have an easy way
                to modify initialization arguments/object attributes without needing to create a
                class that extends the driver, instead allowing the community platforms to simply
                build from the GenericDriver or NetworkDriver classes, and pass this callable to do
                things such as appending to a username (looking at you RouterOS!!). Note that this
                is *always* a synchronous function (even for asyncio drivers)!
            on_open: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately after authentication is completed. Common use
                cases for this callable would be to disable paging or accept any kind of banner
                message that prompts a user upon connection
            on_close: callable that accepts the class instance as its only argument. this callable,
                if provided, is executed immediately prior to closing the underlying transport.
                Common use cases for this callable would be to save configurations prior to exiting,
                or to logout properly to free up vtys or similar
            transport: name of the transport plugin to use for the actual telnet/ssh/netconf
                connection. Available "core" transports are:
                    - system
                    - telnet
                    - asynctelnet
                    - ssh2
                    - paramiko
                    - asyncssh
                Please see relevant transport plugin section for details. Additionally third party
                transport plugins may be available.
            transport_options: dictionary of options to pass to selected transport class; see
                docs for given transport class for details of what to pass here
            channel_lock: True/False to lock the channel (threading.Lock/asyncio.Lock) during
                any channel operations, defaults to False
            channel_log: True/False or a string path to a file of where to write out channel logs --
                these are not "logs" in the normal logging module sense, but only the output that is
                read from the channel. In other words, the output of the channel log should look
                similar to what you would see as a human connecting to a device
            channel_log_mode: "write"|"append", all other values will raise ValueError,
                does what it sounds like it should by setting the channel log to the provided mode
            logging_uid: unique identifier (string) to associate to log messages; useful if you have
                multiple connections to the same device (i.e. one console, one ssh, or one to each
                supervisor module, etc.)
            failed_when_contains: list of strings indicating command/config failure
            textfsm_platform: string to use to fetch ntc-templates templates for textfsm parsing
            genie_platform: string to use to fetch genie parser templates
            privilege_levels: optional user provided privilege levels, if left None will default to
                scrapli standard privilege levels
            default_desired_privilege_level: string of name of default desired priv, this is the
                priv level that is generally used to disable paging/set terminal width and things
                like that upon first login, and is also the priv level scrapli will try to acquire
                for normal "command" operations (`send_command`, `send_commands`)
            auth_secondary: password to use for secondary authentication (enable)
            failed_when_contains: List of strings that indicate a command/config has failed
            variant: name of the community platform variant if desired
            **kwargs: should be unused, but here to accept any additional kwargs from users

        Returns:
            final_driver: synchronous driver class for provided driver

        Raises:
            ScrapliValueError: if provided transport is asyncio
            ScrapliTypeError: if `platform` not in keyword arguments

        """
        logger.debug("Scrapli factory initialized")

        if transport in ASYNCIO_TRANSPORTS:
            raise ScrapliValueError("Use 'AsyncScrapli' if using an async transport!")

        if not isinstance(platform, str):
            raise ScrapliTypeError(f"Argument 'platform' must be 'str' got '{type(platform)}'")

        provided_kwargs = _build_provided_kwargs_dict(
            host=host,
            port=port,
            auth_username=auth_username,
            auth_password=auth_password,
            auth_private_key=auth_private_key,
            auth_private_key_passphrase=auth_private_key_passphrase,
            auth_strict_key=auth_strict_key,
            auth_bypass=auth_bypass,
            timeout_socket=timeout_socket,
            timeout_transport=timeout_transport,
            timeout_ops=timeout_ops,
            comms_return_char=comms_return_char,
            ssh_config_file=ssh_config_file,
            ssh_known_hosts_file=ssh_known_hosts_file,
            on_init=on_init,
            on_open=on_open,
            on_close=on_close,
            transport=transport,
            transport_options=transport_options,
            channel_log=channel_log,
            channel_log_mode=channel_log_mode,
            channel_lock=channel_lock,
            logging_uid=logging_uid,
            privilege_levels=privilege_levels,
            default_desired_privilege_level=default_desired_privilege_level,
            auth_secondary=auth_secondary,
            failed_when_contains=failed_when_contains,
            textfsm_platform=textfsm_platform,
            genie_platform=genie_platform,
            **kwargs,
        )

        final_driver, additional_kwargs = cls._get_driver(platform=platform, variant=variant)

        # at this point will need to merge the additional kwargs in (for community drivers),
        # ensure that kwargs passed by user supersede the ones coming from community platform
        if additional_kwargs:
            final_kwargs = {**additional_kwargs, **provided_kwargs}
        else:
            final_kwargs = provided_kwargs

        final_conn = final_driver(**final_kwargs)
        # cast the final conn to type Scrapli to appease mypy -- we know it will be a NetworkDriver
        # or GenericDriver, but thats ok =)
        final_conn = cast(Scrapli, final_conn)
        return final_conn

__new__(platform: str, host: str, privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None, default_desired_privilege_level: Optional[str] = None, port: Optional[int] = None, auth_username: Optional[str] = None, auth_password: Optional[str] = None, auth_private_key: Optional[str] = None, auth_private_key_passphrase: Optional[str] = None, auth_strict_key: Optional[bool] = None, auth_bypass: Optional[bool] = None, timeout_socket: Optional[float] = None, timeout_transport: Optional[float] = None, timeout_ops: Optional[float] = None, comms_return_char: Optional[str] = None, ssh_config_file: Optional[Union[str, bool]] = None, ssh_known_hosts_file: Optional[Union[str, bool]] = None, on_init: Optional[Callable[..., Any]] = None, on_open: Optional[Callable[..., Any]] = None, on_close: Optional[Callable[..., Any]] = None, transport: Optional[str] = None, transport_options: Optional[Dict[str, Any]] = None, channel_log: Optional[Union[str, bool, BytesIO]] = None, channel_lock: Optional[bool] = None, channel_log_mode: Optional[str] = None, logging_uid: Optional[str] = None, auth_secondary: Optional[str] = None, failed_when_contains: Optional[List[str]] = None, textfsm_platform: Optional[str] = None, genie_platform: Optional[str] = None, variant: Optional[str] = None, **kwargs: Dict[Any, Any]) -> Scrapli

Scrapli Factory method for synchronous drivers

Parameters:

Name	Type	Description	Default
platform	str	name of the scrapli platform to return a connection object for; should be one of the "core" platforms or a valid community platform name	required
host	str	host ip/name to connect to	required
port	Optional[int]	port to connect to	None
auth_username	Optional[str]	username for authentication	None
auth_private_key	Optional[str]	path to private key for authentication	None
auth_private_key_passphrase	Optional[str]	passphrase for decrypting ssh key if necessary	None
auth_password	Optional[str]	password for authentication	None
auth_strict_key	Optional[bool]	strict host checking or not	None
auth_bypass	Optional[bool]	bypass "in channel" authentication -- only supported with telnet, asynctelnet, and system transport plugins	None
timeout_socket	Optional[float]	timeout for establishing socket/initial connection in seconds	None
timeout_transport	Optional[float]	timeout for ssh|telnet transport in seconds	None
timeout_ops	Optional[float]	timeout for ssh channel operations	None
comms_return_char	Optional[str]	character to use to send returns to host	None
ssh_config_file	Optional[Union[str, bool]]	string to path for ssh config file, True to use default ssh config file or False to ignore default ssh config file	None
ssh_known_hosts_file	Optional[Union[str, bool]]	string to path for ssh known hosts file, True to use default known file locations. Only applicable/needed if auth_strict_key is set to True	None
on_init	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed as the last step of object instantiation -- its purpose is primarily to provide a mechanism for scrapli community platforms to have an easy way to modify initialization arguments/object attributes without needing to create a class that extends the driver, instead allowing the community platforms to simply build from the GenericDriver or NetworkDriver classes, and pass this callable to do things such as appending to a username (looking at you RouterOS!!). Note that this is always a synchronous function (even for asyncio drivers)!	None
on_open	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed immediately after authentication is completed. Common use cases for this callable would be to disable paging or accept any kind of banner message that prompts a user upon connection	None
on_close	Optional[Callable[..., Any]]	callable that accepts the class instance as its only argument. this callable, if provided, is executed immediately prior to closing the underlying transport. Common use cases for this callable would be to save configurations prior to exiting, or to logout properly to free up vtys or similar	None
transport	Optional[str]	name of the transport plugin to use for the actual telnet/ssh/netconf connection. Available "core" transports are: - system - telnet - asynctelnet - ssh2 - paramiko - asyncssh Please see relevant transport plugin section for details. Additionally third party transport plugins may be available.	None
transport_options	Optional[Dict[str, Any]]	dictionary of options to pass to selected transport class; see docs for given transport class for details of what to pass here	None
channel_lock	Optional[bool]	True/False to lock the channel (threading.Lock/asyncio.Lock) during any channel operations, defaults to False	None
channel_log	Optional[Union[str, bool, BytesIO]]	True/False or a string path to a file of where to write out channel logs -- these are not "logs" in the normal logging module sense, but only the output that is read from the channel. In other words, the output of the channel log should look similar to what you would see as a human connecting to a device	None
channel_log_mode	Optional[str]	"write"|"append", all other values will raise ValueError, does what it sounds like it should by setting the channel log to the provided mode	None
logging_uid	Optional[str]	unique identifier (string) to associate to log messages; useful if you have multiple connections to the same device (i.e. one console, one ssh, or one to each supervisor module, etc.)	None
failed_when_contains	Optional[List[str]]	list of strings indicating command/config failure	None
textfsm_platform	Optional[str]	string to use to fetch ntc-templates templates for textfsm parsing	None
genie_platform	Optional[str]	string to use to fetch genie parser templates	None
privilege_levels	Optional[Dict[str, PrivilegeLevel]]	optional user provided privilege levels, if left None will default to scrapli standard privilege levels	None
default_desired_privilege_level	Optional[str]	string of name of default desired priv, this is the priv level that is generally used to disable paging/set terminal width and things like that upon first login, and is also the priv level scrapli will try to acquire for normal "command" operations (send_command, send_commands)	None
auth_secondary	Optional[str]	password to use for secondary authentication (enable)	None
failed_when_contains	Optional[List[str]]	List of strings that indicate a command/config has failed	None
variant	Optional[str]	name of the community platform variant if desired	None
**kwargs	Dict[Any, Any]	should be unused, but here to accept any additional kwargs from users	{}

Returns:

Name	Type	Description
final_driver	Scrapli	synchronous driver class for provided driver

Raises:

Type	Description
ScrapliValueError	if provided transport is asyncio
ScrapliTypeError	if platform not in keyword arguments

Source code in scrapli/factory.py

def __new__(  # pylint: disable=R0914
    cls,
    platform: str,
    host: str,
    privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
    default_desired_privilege_level: Optional[str] = None,
    port: Optional[int] = None,
    auth_username: Optional[str] = None,
    auth_password: Optional[str] = None,
    auth_private_key: Optional[str] = None,
    auth_private_key_passphrase: Optional[str] = None,
    auth_strict_key: Optional[bool] = None,
    auth_bypass: Optional[bool] = None,
    timeout_socket: Optional[float] = None,
    timeout_transport: Optional[float] = None,
    timeout_ops: Optional[float] = None,
    comms_return_char: Optional[str] = None,
    ssh_config_file: Optional[Union[str, bool]] = None,
    ssh_known_hosts_file: Optional[Union[str, bool]] = None,
    on_init: Optional[Callable[..., Any]] = None,
    on_open: Optional[Callable[..., Any]] = None,
    on_close: Optional[Callable[..., Any]] = None,
    transport: Optional[str] = None,
    transport_options: Optional[Dict[str, Any]] = None,
    channel_log: Optional[Union[str, bool, BytesIO]] = None,
    channel_lock: Optional[bool] = None,
    channel_log_mode: Optional[str] = None,
    logging_uid: Optional[str] = None,
    auth_secondary: Optional[str] = None,
    failed_when_contains: Optional[List[str]] = None,
    textfsm_platform: Optional[str] = None,
    genie_platform: Optional[str] = None,
    variant: Optional[str] = None,
    **kwargs: Dict[Any, Any],
) -> "Scrapli":
    r"""
    Scrapli Factory method for synchronous drivers

    Args:
        platform: name of the scrapli platform to return a connection object for; should be
            one of the "core" platforms or a valid community platform name
        host: host ip/name to connect to
        port: port to connect to
        auth_username: username for authentication
        auth_private_key: path to private key for authentication
        auth_private_key_passphrase: passphrase for decrypting ssh key if necessary
        auth_password: password for authentication
        auth_strict_key: strict host checking or not
        auth_bypass: bypass "in channel" authentication -- only supported with telnet,
            asynctelnet, and system transport plugins
        timeout_socket: timeout for establishing socket/initial connection in seconds
        timeout_transport: timeout for ssh|telnet transport in seconds
        timeout_ops: timeout for ssh channel operations
        comms_return_char: character to use to send returns to host
        ssh_config_file: string to path for ssh config file, True to use default ssh config file
            or False to ignore default ssh config file
        ssh_known_hosts_file: string to path for ssh known hosts file, True to use default known
            file locations. Only applicable/needed if `auth_strict_key` is set to True
        on_init: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed as the last step of object instantiation -- its purpose is
            primarily to provide a mechanism for scrapli community platforms to have an easy way
            to modify initialization arguments/object attributes without needing to create a
            class that extends the driver, instead allowing the community platforms to simply
            build from the GenericDriver or NetworkDriver classes, and pass this callable to do
            things such as appending to a username (looking at you RouterOS!!). Note that this
            is *always* a synchronous function (even for asyncio drivers)!
        on_open: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately after authentication is completed. Common use
            cases for this callable would be to disable paging or accept any kind of banner
            message that prompts a user upon connection
        on_close: callable that accepts the class instance as its only argument. this callable,
            if provided, is executed immediately prior to closing the underlying transport.
            Common use cases for this callable would be to save configurations prior to exiting,
            or to logout properly to free up vtys or similar
        transport: name of the transport plugin to use for the actual telnet/ssh/netconf
            connection. Available "core" transports are:
                - system
                - telnet
                - asynctelnet
                - ssh2
                - paramiko
                - asyncssh
            Please see relevant transport plugin section for details. Additionally third party
            transport plugins may be available.
        transport_options: dictionary of options to pass to selected transport class; see
            docs for given transport class for details of what to pass here
        channel_lock: True/False to lock the channel (threading.Lock/asyncio.Lock) during
            any channel operations, defaults to False
        channel_log: True/False or a string path to a file of where to write out channel logs --
            these are not "logs" in the normal logging module sense, but only the output that is
            read from the channel. In other words, the output of the channel log should look
            similar to what you would see as a human connecting to a device
        channel_log_mode: "write"|"append", all other values will raise ValueError,
            does what it sounds like it should by setting the channel log to the provided mode
        logging_uid: unique identifier (string) to associate to log messages; useful if you have
            multiple connections to the same device (i.e. one console, one ssh, or one to each
            supervisor module, etc.)
        failed_when_contains: list of strings indicating command/config failure
        textfsm_platform: string to use to fetch ntc-templates templates for textfsm parsing
        genie_platform: string to use to fetch genie parser templates
        privilege_levels: optional user provided privilege levels, if left None will default to
            scrapli standard privilege levels
        default_desired_privilege_level: string of name of default desired priv, this is the
            priv level that is generally used to disable paging/set terminal width and things
            like that upon first login, and is also the priv level scrapli will try to acquire
            for normal "command" operations (`send_command`, `send_commands`)
        auth_secondary: password to use for secondary authentication (enable)
        failed_when_contains: List of strings that indicate a command/config has failed
        variant: name of the community platform variant if desired
        **kwargs: should be unused, but here to accept any additional kwargs from users

    Returns:
        final_driver: synchronous driver class for provided driver

    Raises:
        ScrapliValueError: if provided transport is asyncio
        ScrapliTypeError: if `platform` not in keyword arguments

    """
    logger.debug("Scrapli factory initialized")

    if transport in ASYNCIO_TRANSPORTS:
        raise ScrapliValueError("Use 'AsyncScrapli' if using an async transport!")

    if not isinstance(platform, str):
        raise ScrapliTypeError(f"Argument 'platform' must be 'str' got '{type(platform)}'")

    provided_kwargs = _build_provided_kwargs_dict(
        host=host,
        port=port,
        auth_username=auth_username,
        auth_password=auth_password,
        auth_private_key=auth_private_key,
        auth_private_key_passphrase=auth_private_key_passphrase,
        auth_strict_key=auth_strict_key,
        auth_bypass=auth_bypass,
        timeout_socket=timeout_socket,
        timeout_transport=timeout_transport,
        timeout_ops=timeout_ops,
        comms_return_char=comms_return_char,
        ssh_config_file=ssh_config_file,
        ssh_known_hosts_file=ssh_known_hosts_file,
        on_init=on_init,
        on_open=on_open,
        on_close=on_close,
        transport=transport,
        transport_options=transport_options,
        channel_log=channel_log,
        channel_log_mode=channel_log_mode,
        channel_lock=channel_lock,
        logging_uid=logging_uid,
        privilege_levels=privilege_levels,
        default_desired_privilege_level=default_desired_privilege_level,
        auth_secondary=auth_secondary,
        failed_when_contains=failed_when_contains,
        textfsm_platform=textfsm_platform,
        genie_platform=genie_platform,
        **kwargs,
    )

    final_driver, additional_kwargs = cls._get_driver(platform=platform, variant=variant)

    # at this point will need to merge the additional kwargs in (for community drivers),
    # ensure that kwargs passed by user supersede the ones coming from community platform
    if additional_kwargs:
        final_kwargs = {**additional_kwargs, **provided_kwargs}
    else:
        final_kwargs = provided_kwargs

    final_conn = final_driver(**final_kwargs)
    # cast the final conn to type Scrapli to appease mypy -- we know it will be a NetworkDriver
    # or GenericDriver, but thats ok =)
    final_conn = cast(Scrapli, final_conn)
    return final_conn