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

Initializes 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 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 to wait for lab to converge.
convergence_wait_time – wait interval in seconds to wait during one iteration during lab convergence wait.
events – whether to use events - event listeners for state updates and event handlers for custom handling.

Raises:

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

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

Joins all labs owned by this user (or all labs period) and returns 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() → None: Checks remote controller version against current client version (specified in self.VERSION) and against controller version blacklist (specified in self.INCOMPATIBLE_CONTROLLER_VERSIONS) and raises exception if versions are incompatible or prints warning if client minor version is lower than controller minor version.

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

Creates an empty lab with the given name. If no name is given, then the created lab ID is set as the name.

Note that the lab will be auto-syncing according to the Client Library’s auto-sync setting when created. The lab has a property to override this on a per-lab basis.

Example:

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

Parameters:: title – The Lab name (or title)
Returns:: A Lab instance

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 which match the title specified

get_diagnostics() → dict

Returns the controller diagnostic data as JSON

Returns:: diagnostic data

get_host() → str

Returns the hostname of the session to the controller.

Returns:: A hostname

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

Returns list of all lab IDs.

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

get_local_lab(lab_id: str) → Lab

get_sample_labs() → dict[str, dict]

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

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

get_system_health() → dict

Returns the controller system health data as JSON

Returns:: system health data

get_system_stats() → dict

Returns the controller resource statistics as JSON

Returns:: system resource statistics

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

Imports an existing topology from a string.

Parameters:

topology – Topology representation as a string
title – Title of the lab (optional)
offline – whether the ClientLibrary should import the lab locally. The topology parameter has to be a JSON string in this case. This can not be XML or YAML representation of the lab. Compatible JSON from GET /labs/{lab_id}/topology.
virl_1x – Is this old virl-1x topology format (default=False)

Returns:

A Lab instance

Raises:

ValueError – if there’s no lab ID in the API response
httpx.HTTPError – if there was a transport error

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

Imports an existing topology from a file / path.

Parameters:

path – Topology filename / path
title – Title of the lab

Returns:

A Lab instance

import_sample_lab(title: str) → Lab

Imports a built-in sample lab.

Parameters:: title – sample lab name, as returned by get_sample_labs
Returns:: a Lab instance

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

Reports whether the system is ready or not.

Parameters:

wait – if this is true, the call will block until it’s ready
max_wait – maximum time to wait, in seconds
sleep – time to wait between tries, in seconds

Returns:

ready state

static is_virl_1x(path: Path) → bool

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

Creates a new ClientLibrary lab instance that is retrieved from the controller.

If sync_lab is set to true, then synchronize current lab by applying the changes that were done in UI or in another ClientLibrary session.

Join preexisting lab:

lab = client_library.join_existing_lab("2e6a18")

Parameters:

lab_id – The lab ID to be joined.
sync_lab – Synchronize changes.

Returns:

A Lab instance

local_labs() → list[Lab]

logout(clear_all_sessions: bool = False) → None: Invalidate current token.

remove_lab(lab_id: str | Lab) → None

Use carefully, it removes a given lab:

client_library.remove_lab("1f6cd7")

If you have a lab object, you can also simply do:

lab.remove()

Parameters:: lab_id – the lab ID or object

start_event_listening()

Starts listening for and parsing websocket events.

To replace the default event handling mechanism, subclass event_handling.EventHandler (or EventHandlerBase if necessary), then do:

self.websockets = event_listening.EventListener(self) self.websockets._event_handler = handler

stop_event_listening()

system_info() → dict

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

Returns:: system information json

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.6.1

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 session: Client

Returns the used httpx client session object.

Returns:: The session object

property uuid: str

Returns the UUID4 that identifies this client to the server.

Returns:: the UUID4 string

Subpackages

virl2_client.models package

Submodules

virl2_client.event_handling

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

Bases: object

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

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

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.

parse_event(event: Event) → None

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

Bases: ABC

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

__init__(client_library: ClientLibrary = None)

parse_event(event: Event) → None

virl2_client.event_listening

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

Bases: object

__init__(client_library: ClientLibrary)

Initializes a 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()

stop_listening()

virl2_client.exceptions

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.InvalidContentType: Bases: VirlException

exception virl2_client.exceptions.InvalidImageFile: Bases: VirlException

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.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.

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

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

Initializes 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 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 to wait for lab to converge.
convergence_wait_time – wait interval in seconds to wait during one iteration during lab convergence wait.
events – whether to use events - event listeners for state updates and event handlers for custom handling.

Raises:

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

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

Joins all labs owned by this user (or all labs period) and returns 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() → None: Checks remote controller version against current client version (specified in self.VERSION) and against controller version blacklist (specified in self.INCOMPATIBLE_CONTROLLER_VERSIONS) and raises exception if versions are incompatible or prints warning if client minor version is lower than controller minor version.

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

Creates an empty lab with the given name. If no name is given, then the created lab ID is set as the name.

Note that the lab will be auto-syncing according to the Client Library’s auto-sync setting when created. The lab has a property to override this on a per-lab basis.

Example:

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

Parameters:: title – The Lab name (or title)
Returns:: A Lab instance

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 which match the title specified

get_diagnostics() → dict

Returns the controller diagnostic data as JSON

Returns:: diagnostic data

get_host() → str

Returns the hostname of the session to the controller.

Returns:: A hostname

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

Returns list of all lab IDs.

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

get_local_lab(lab_id: str) → Lab

get_sample_labs() → dict[str, dict]

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

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

get_system_health() → dict

Returns the controller system health data as JSON

Returns:: system health data

get_system_stats() → dict

Returns the controller resource statistics as JSON

Returns:: system resource statistics

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

Imports an existing topology from a string.

Parameters:

topology – Topology representation as a string
title – Title of the lab (optional)
offline – whether the ClientLibrary should import the lab locally. The topology parameter has to be a JSON string in this case. This can not be XML or YAML representation of the lab. Compatible JSON from GET /labs/{lab_id}/topology.
virl_1x – Is this old virl-1x topology format (default=False)

Returns:

A Lab instance

Raises:

ValueError – if there’s no lab ID in the API response
httpx.HTTPError – if there was a transport error

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

Imports an existing topology from a file / path.

Parameters:

path – Topology filename / path
title – Title of the lab

Returns:

A Lab instance

import_sample_lab(title: str) → Lab

Imports a built-in sample lab.

Parameters:: title – sample lab name, as returned by get_sample_labs
Returns:: a Lab instance

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

Reports whether the system is ready or not.

Parameters:

wait – if this is true, the call will block until it’s ready
max_wait – maximum time to wait, in seconds
sleep – time to wait between tries, in seconds

Returns:

ready state

static is_virl_1x(path: Path) → bool

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

Creates a new ClientLibrary lab instance that is retrieved from the controller.

If sync_lab is set to true, then synchronize current lab by applying the changes that were done in UI or in another ClientLibrary session.

Join preexisting lab:

lab = client_library.join_existing_lab("2e6a18")

Parameters:

lab_id – The lab ID to be joined.
sync_lab – Synchronize changes.

Returns:

A Lab instance

local_labs() → list[Lab]

logout(clear_all_sessions: bool = False) → None: Invalidate current token.

remove_lab(lab_id: str | Lab) → None

Use carefully, it removes a given lab:

client_library.remove_lab("1f6cd7")

If you have a lab object, you can also simply do:

lab.remove()

Parameters:: lab_id – the lab ID or object

start_event_listening()

Starts listening for and parsing websocket events.

To replace the default event handling mechanism, subclass event_handling.EventHandler (or EventHandlerBase if necessary), then do:

self.websockets = event_listening.EventListener(self) self.websockets._event_handler = handler

stop_event_listening()

system_info() → dict

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

Returns:: system information json

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.6.1

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

property session: Client

Returns the used httpx client session object.

Returns:: The session object

url: str

username: str

property uuid: str

Returns the UUID4 that identifies this client to the server.

Returns:: the UUID4 string

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

virl2_client.virl2_client.prepare_url(url: str, allow_http: bool) → tuple[str, str]