virl2_client package

exception virl2_client.InitializationError: Bases: VirlException

exception virl2_client.InterfaceNotFound: Bases: ElementNotFound

exception virl2_client.LabNotFound: Bases: ElementNotFound

exception virl2_client.LinkNotFound: Bases: ElementNotFound

exception virl2_client.NodeNotFound: Bases: ElementNotFound

class virl2_client.ClientConfig(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, allow_http: bool = False, auto_sync: float = 1.0, events: bool = False, raise_for_auth_failure: bool = True, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5)

Bases: tuple

Stores client library configuration, which can be used to create any number of identically configured instances of ClientLibrary.

make_client() → ClientLibrary

allow_http: bool: Alias for field number 4

auto_sync: float: Alias for field number 5

convergence_wait_max_iter: int: Alias for field number 8

convergence_wait_time: int | float: Alias for field number 9

events: bool: Alias for field number 6

password: str | None: Alias for field number 2

raise_for_auth_failure: bool: Alias for field number 7

ssl_verify: bool | str: Alias for field number 3

url: str | None: Alias for field number 0

username: str | None: Alias for field number 1

class virl2_client.ClientLibrary(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, raise_for_auth_failure: bool = False, allow_http: bool = False, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5, events: bool = False)

Bases: object

Python bindings for the REST API of a CML controller.

__init__(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, raise_for_auth_failure: bool = False, allow_http: bool = False, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5, events: bool = False) → None

Initialize a ClientLibrary instance. Note that ssl_verify can also be a string that points to a cert (see class documentation).

Parameters:

url – URL of controller. It’s also possible to pass the URL via the VIRL2_URL or VIRL_HOST environment variable. If no protocol scheme is provided, “https:” is used.
username – Username of the user to authenticate. It’s also possible to pass the username via VIRL2_USER or VIRL_USERNAME variable.
password – Password of the user to authenticate. It’s also possible to pass the password via VIRL2_PASS or VIRL_PASSWORD variable.
ssl_verify – Path of the SSL controller certificate, or True to load from CA_BUNDLE or CML_VERIFY_CERT environment variable, or False to disable.
raise_for_auth_failure – Raise an exception if unable to connect to controller. (Use for scripting scenarios.)
allow_http – If set, a https URL will not be enforced.
convergence_wait_max_iter – Maximum number of iterations for convergence.
convergence_wait_time – Time in seconds to sleep between convergence calls on the backend.
events – A flag indicating whether to enable event-based data synchronization from the server. When enabled, utilizes a mechanism for receiving real-time updates from the server, instead of periodically requesting the data.

Raises:

InitializationError – If no URL is provided, authentication fails or host can’t be reached.

all_labs(show_all: bool = False) → list[Lab]

Join all labs owned by this user (or all labs) and return their list.

Parameters:: show_all – Whether to get only labs owned by the admin or all user labs.
Returns:: A list of Lab objects.

check_controller_version() → Version | None: Check remote controller version against current client version (specified in self.VERSION) and against controller version blacklist (specified in self.INCOMPATIBLE_CONTROLLER_VERSIONS). Raise exception if versions are incompatible, or print warning if the client minor version is lower than the controller minor version.

create_lab(title: str | None = None, description: str | None = None, notes: str | None = None) → Lab

Create a new lab with optional title, description, and notes.

If no title, description, or notes are provided, the server will generate a default title in the format “Lab at Mon 13:30 PM” and leave the description and notes blank.

The lab will automatically sync based on the Client Library’s auto-sync setting when created, but this behavior can be overridden on a per-lab basis.

Example:

lab = client_library.create_lab()
print(lab.id)
lab.create_node("r1", "iosv", 50, 100)

Parameters:

title – The title of the lab.
description – The description of the lab.
notes – The notes of the lab.

Returns:

A Lab instance representing the created lab.

find_labs_by_title(title: str) → list[Lab]

Return a list of labs which match the given title.

Parameters:: title – The title to search for.
Returns:: A list of Lab objects matching the specified title.

get_diagnostics() → dict

Return the controller diagnostic data as a JSON object.

Returns:: The diagnostic data.

get_host() → str

Return the hostname of the session to the controller.

Returns:: The hostname.

get_lab_list(show_all: bool = False) → list[str]

Get the list of all lab IDs.

Parameters:: show_all – Whether to include labs owned by all users (True) or only labs owned by the admin (False).
Returns:: A list of lab IDs.

get_local_lab(lab_id: str) → Lab

Get a local lab by its ID.

Parameters:: lab_id – The ID of the lab.
Returns:: The Lab object with the specified ID.
Raises:: LabNotFound – If the lab with the given ID does not exist.

get_sample_labs() → dict[str, dict]

Return a dictionary with information about all sample labs available on the host.

Returns:: A dictionary of sample lab information where the keys are the titles.

get_system_health() → dict

Return the controller system health data as a JSON object.

Returns:: The system health data.

get_system_stats() → dict

Return the controller resource statistics as a JSON object.

Returns:: The system resource statistics.

import_lab(topology: str, title: str | None = None, offline: bool = False, virl_1x: bool = False) → Lab

Import an existing topology from a string.

Parameters:

topology – The topology representation as a string.
title – The title of the lab.
offline – Whether to import the lab locally.
virl_1x – Whether the topology format is the old, VIRL 1.x format.

Returns:

The imported Lab instance.

Raises:

ValueError – If no lab ID is returned in the API response.
httpx.HTTPError – If there was a transport error.

import_lab_from_path(path: str, title: str | None = None) → Lab

Import an existing topology from a file or path.

Parameters:

path – The topology filename or path.
title – The title of the lab.

Returns:

The imported Lab instance.

Raises:

FileNotFoundError – If the specified path does not exist.

import_sample_lab(title: str) → Lab

Import a built-in sample lab.

Parameters:: title – The sample lab name.
Returns:: The imported Lab instance.

is_system_ready(wait: bool = False, max_wait: int = 60, sleep: int = 5) → bool

Report whether the system is ready or not.

Parameters:

wait – Whether to block until the system is ready.
max_wait – The maximum time to wait in seconds.
sleep – The time to wait between tries in seconds.

Returns:

The ready state of the system.

static is_virl_1x(path: Path) → bool

Check if the given file is of VIRL version 1.x.

Parameters:: path – The path to check.
Returns:: Whether the file is of VIRL version 1.x.

join_existing_lab(lab_id: str, sync_lab: bool = True) → Lab

Join a lab that exists on the server and make it accessible locally.

If sync_lab is set to True, the current lab will be synchronized by applying any changes that were made in the UI or in another ClientLibrary session.

Example::: lab = client_library.join_existing_lab(“2e6a18”)

Parameters:

lab_id – The ID of the lab to be joined.
sync_lab – Whether to synchronize the lab.

Returns:

A Lab instance representing the joined lab.

Raises:

LabNotFound – If no lab with the given ID exists on the host.

local_labs() → list[Lab]

Return a list of local labs.

Returns:: A list of local labs.

logout(clear_all_sessions: bool = False) → bool

Invalidate the current token.

Parameters:: clear_all_sessions – Whether to clear all user sessions as well.

remove_lab(lab_id: str | Lab) → None

Remove a lab identified by its ID or Lab object.

Use this method with caution as it permanently deletes the specified lab.

If you have the lab object, you can also do lab.remove().

Parameters:: lab_id – The ID or Lab object representing the lab to be removed.

start_event_listening()

Start listening for and parsing websocket events.

To replace the default event handling mechanism, subclass event_handling.EventHandler (or EventHandlerBase if necessary), then do:: from .event_listening import EventListener custom_listener = EventListener() custom_listener._event_handler = CustomHandler(client_library) client_library.event_listener = custom_listener

Returns:

stop_event_listening(): Stop listening for and parsing websocket events.

system_info() → dict

Get information about the system where the application runs. Can be called without authentication.

Returns:: The system information as a JSON object.

INCOMPATIBLE_CONTROLLER_VERSIONS = [2.0.0, 2.0.1, 2.1.0, 2.1.1, 2.1.2, 2.2.1, 2.2.2, 2.2.3]

VERSION = 2.7.0

auto_sync: auto_sync automatically syncs data with the backend after a specific time. The default expiry time is 1.0s. This time can be configured by setting the auto_sync_interval.

property uuid: str: Return the UUID4 that identifies this client to the server.

Subpackages

virl2_client.models package

Submodules

virl2_client.event_handling

class virl2_client.event_handling.Event(event_dict: dict[str, Any])

Bases: object

__init__(event_dict: dict[str, Any])

An event object, stores parsed info about the event it represents.

Parameters:: event_dict – A dictionary containing event data.

class virl2_client.event_handling.EventHandler(client_library: ClientLibrary = None)

Bases: EventHandlerBase

Handler for JSON events received from controller over websockets. Used by EventListener by default, but can be subclassed and methods overridden to change/extend the handling mechanism, then passed to EventListener.

handle_event(event: Event) → None

Parse and handle the given event.

Parameters:: event – An Event object representing the event to be parsed.

class virl2_client.event_handling.EventHandlerBase(client_library: ClientLibrary = None)

Bases: ABC

__init__(client_library: ClientLibrary = None)

Abstract base class for event handlers.

Subclass this if you want to implement the entire event handling mechanism from scratch, otherwise subclass EventHandler instead.

Parameters:: client_library – The client library which should be modified by events.

handle_event(event: Event) → None

Parse and handle the given event.

Parameters:: event – An Event object representing the event to be parsed.

virl2_client.event_listening

class virl2_client.event_listening.EventListener(client_library: ClientLibrary)

Bases: object

__init__(client_library: ClientLibrary)

Initialize an EventListener instance. EventListener creates and listens to a websocket connection to the server. Events are then sent to the EventHandler instance for handling. Use start_listening() to open and stop_listening() to close connection.

Parameters:: client_library – Parent ClientLibrary instance which provides connection info and is modified when synchronizing.

start_listening(): Start listening for events.

stop_listening(): Stop listening for events.

virl2_client.exceptions

exception virl2_client.exceptions.AnnotationNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.ControllerNotFound

Bases: VirlException

message = 'Controller not found'

exception virl2_client.exceptions.DesynchronizedError: Bases: VirlException

exception virl2_client.exceptions.ElementAlreadyExists: Bases: VirlException, FileExistsError

exception virl2_client.exceptions.ElementNotFound: Bases: VirlException, KeyError

exception virl2_client.exceptions.InitializationError: Bases: VirlException

exception virl2_client.exceptions.InterfaceNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.InvalidAnnotationType: Bases: VirlException

exception virl2_client.exceptions.InvalidContentType: Bases: VirlException

exception virl2_client.exceptions.InvalidImageFile: Bases: VirlException

exception virl2_client.exceptions.InvalidMacAddressBlock

Bases: VirlException

message = 'MAC address block has to be in range 0-7'

exception virl2_client.exceptions.InvalidProperty: Bases: VirlException

exception virl2_client.exceptions.LabNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.LinkNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.MethodNotActive: Bases: VirlException

exception virl2_client.exceptions.NodeNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.PyatsDeviceNotFound: Bases: PyatsException

exception virl2_client.exceptions.PyatsException: Bases: Exception

exception virl2_client.exceptions.PyatsNotInstalled: Bases: PyatsException

exception virl2_client.exceptions.VirlException: Bases: Exception

virl2_client.utils

class virl2_client.utils.property_s(fget=None, fset=None, fdel=None, doc=None)

Bases: property

A modified property that will check staleness.

__init__(fget=None, fset=None, fdel=None, doc=None)

virl2_client.utils.check_stale(func: TCallable) → TCallable: A decorator that will make the wrapped function check staleness.

virl2_client.utils.get_url_from_template(endpoint: str, url_templates: dict[str, str], values: dict | None = None) → str

Generate the URL for a given API endpoint from given templates.

Parameters:

endpoint – The desired endpoint.
url_templates – The templates to map values to.
values – Keyword arguments used to format the URL.

Returns:

The formatted URL.

virl2_client.utils.locked(func: TCallable) → TCallable: A decorator that makes a method threadsafe. Parent class instance must have a session.lock property for locking to occur.

virl2_client.virl2_client

class virl2_client.virl2_client.ClientConfig(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, allow_http: bool = False, auto_sync: float = 1.0, events: bool = False, raise_for_auth_failure: bool = True, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5)

Bases: tuple

Stores client library configuration, which can be used to create any number of identically configured instances of ClientLibrary.

make_client() → ClientLibrary

allow_http: bool: Alias for field number 4

auto_sync: float: Alias for field number 5

convergence_wait_max_iter: int: Alias for field number 8

convergence_wait_time: int | float: Alias for field number 9

events: bool: Alias for field number 6

password: str | None: Alias for field number 2

raise_for_auth_failure: bool: Alias for field number 7

ssl_verify: bool | str: Alias for field number 3

url: str | None: Alias for field number 0

username: str | None: Alias for field number 1

class virl2_client.virl2_client.ClientLibrary(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, raise_for_auth_failure: bool = False, allow_http: bool = False, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5, events: bool = False)

Bases: object

Python bindings for the REST API of a CML controller.

__init__(url: str | None = None, username: str | None = None, password: str | None = None, ssl_verify: bool | str = True, raise_for_auth_failure: bool = False, allow_http: bool = False, convergence_wait_max_iter: int = 500, convergence_wait_time: int | float = 5, events: bool = False) → None

Initialize a ClientLibrary instance. Note that ssl_verify can also be a string that points to a cert (see class documentation).

Parameters:

url – URL of controller. It’s also possible to pass the URL via the VIRL2_URL or VIRL_HOST environment variable. If no protocol scheme is provided, “https:” is used.
username – Username of the user to authenticate. It’s also possible to pass the username via VIRL2_USER or VIRL_USERNAME variable.
password – Password of the user to authenticate. It’s also possible to pass the password via VIRL2_PASS or VIRL_PASSWORD variable.
ssl_verify – Path of the SSL controller certificate, or True to load from CA_BUNDLE or CML_VERIFY_CERT environment variable, or False to disable.
raise_for_auth_failure – Raise an exception if unable to connect to controller. (Use for scripting scenarios.)
allow_http – If set, a https URL will not be enforced.
convergence_wait_max_iter – Maximum number of iterations for convergence.
convergence_wait_time – Time in seconds to sleep between convergence calls on the backend.
events – A flag indicating whether to enable event-based data synchronization from the server. When enabled, utilizes a mechanism for receiving real-time updates from the server, instead of periodically requesting the data.

Raises:

InitializationError – If no URL is provided, authentication fails or host can’t be reached.

all_labs(show_all: bool = False) → list[Lab]

Join all labs owned by this user (or all labs) and return their list.

Parameters:: show_all – Whether to get only labs owned by the admin or all user labs.
Returns:: A list of Lab objects.

check_controller_version() → Version | None: Check remote controller version against current client version (specified in self.VERSION) and against controller version blacklist (specified in self.INCOMPATIBLE_CONTROLLER_VERSIONS). Raise exception if versions are incompatible, or print warning if the client minor version is lower than the controller minor version.

create_lab(title: str | None = None, description: str | None = None, notes: str | None = None) → Lab

Create a new lab with optional title, description, and notes.

If no title, description, or notes are provided, the server will generate a default title in the format “Lab at Mon 13:30 PM” and leave the description and notes blank.

The lab will automatically sync based on the Client Library’s auto-sync setting when created, but this behavior can be overridden on a per-lab basis.

Example:

lab = client_library.create_lab()
print(lab.id)
lab.create_node("r1", "iosv", 50, 100)

Parameters:

title – The title of the lab.
description – The description of the lab.
notes – The notes of the lab.

Returns:

A Lab instance representing the created lab.

find_labs_by_title(title: str) → list[Lab]

Return a list of labs which match the given title.

Parameters:: title – The title to search for.
Returns:: A list of Lab objects matching the specified title.

get_diagnostics() → dict

Return the controller diagnostic data as a JSON object.

Returns:: The diagnostic data.

get_host() → str

Return the hostname of the session to the controller.

Returns:: The hostname.

get_lab_list(show_all: bool = False) → list[str]

Get the list of all lab IDs.

Parameters:: show_all – Whether to include labs owned by all users (True) or only labs owned by the admin (False).
Returns:: A list of lab IDs.

get_local_lab(lab_id: str) → Lab

Get a local lab by its ID.

Parameters:: lab_id – The ID of the lab.
Returns:: The Lab object with the specified ID.
Raises:: LabNotFound – If the lab with the given ID does not exist.

get_sample_labs() → dict[str, dict]

Return a dictionary with information about all sample labs available on the host.

Returns:: A dictionary of sample lab information where the keys are the titles.

get_system_health() → dict

Return the controller system health data as a JSON object.

Returns:: The system health data.

get_system_stats() → dict

Return the controller resource statistics as a JSON object.

Returns:: The system resource statistics.

import_lab(topology: str, title: str | None = None, offline: bool = False, virl_1x: bool = False) → Lab

Import an existing topology from a string.

Parameters:

topology – The topology representation as a string.
title – The title of the lab.
offline – Whether to import the lab locally.
virl_1x – Whether the topology format is the old, VIRL 1.x format.

Returns:

The imported Lab instance.

Raises:

ValueError – If no lab ID is returned in the API response.
httpx.HTTPError – If there was a transport error.

import_lab_from_path(path: str, title: str | None = None) → Lab

Import an existing topology from a file or path.

Parameters:

path – The topology filename or path.
title – The title of the lab.

Returns:

The imported Lab instance.

Raises:

FileNotFoundError – If the specified path does not exist.

import_sample_lab(title: str) → Lab

Import a built-in sample lab.

Parameters:: title – The sample lab name.
Returns:: The imported Lab instance.

is_system_ready(wait: bool = False, max_wait: int = 60, sleep: int = 5) → bool

Report whether the system is ready or not.

Parameters:

wait – Whether to block until the system is ready.
max_wait – The maximum time to wait in seconds.
sleep – The time to wait between tries in seconds.

Returns:

The ready state of the system.

static is_virl_1x(path: Path) → bool

Check if the given file is of VIRL version 1.x.

Parameters:: path – The path to check.
Returns:: Whether the file is of VIRL version 1.x.

join_existing_lab(lab_id: str, sync_lab: bool = True) → Lab

Join a lab that exists on the server and make it accessible locally.

If sync_lab is set to True, the current lab will be synchronized by applying any changes that were made in the UI or in another ClientLibrary session.

Example::: lab = client_library.join_existing_lab(“2e6a18”)

Parameters:

lab_id – The ID of the lab to be joined.
sync_lab – Whether to synchronize the lab.

Returns:

A Lab instance representing the joined lab.

Raises:

LabNotFound – If no lab with the given ID exists on the host.

local_labs() → list[Lab]

Return a list of local labs.

Returns:: A list of local labs.

logout(clear_all_sessions: bool = False) → bool

Invalidate the current token.

Parameters:: clear_all_sessions – Whether to clear all user sessions as well.

remove_lab(lab_id: str | Lab) → None

Remove a lab identified by its ID or Lab object.

Use this method with caution as it permanently deletes the specified lab.

If you have the lab object, you can also do lab.remove().

Parameters:: lab_id – The ID or Lab object representing the lab to be removed.

start_event_listening()

Start listening for and parsing websocket events.

To replace the default event handling mechanism, subclass event_handling.EventHandler (or EventHandlerBase if necessary), then do:: from .event_listening import EventListener custom_listener = EventListener() custom_listener._event_handler = CustomHandler(client_library) client_library.event_listener = custom_listener

Returns:

stop_event_listening(): Stop listening for and parsing websocket events.

system_info() → dict

Get information about the system where the application runs. Can be called without authentication.

Returns:: The system information as a JSON object.

INCOMPATIBLE_CONTROLLER_VERSIONS = [2.0.0, 2.0.1, 2.1.0, 2.1.1, 2.1.2, 2.2.1, 2.2.2, 2.2.3]

VERSION = 2.7.0

auto_sync: auto_sync automatically syncs data with the backend after a specific time. The default expiry time is 1.0s. This time can be configured by setting the auto_sync_interval.

password: str

url: str

username: str

property uuid: str: Return the UUID4 that identifies this client to the server.

class virl2_client.virl2_client.Version(version_str: str)

Bases: object

__init__(version_str: str) → None

major_differs(other: Version) → bool

minor_differs(other: Version) → bool

minor_lt(other: Version) → bool

minor_or_patch_differs(other: Version) → bool

static parse_version_str(version_str: str) → str

patch_differs(other: Version) → bool

major

minor

patch

version_str