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, jwtoken: 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, check_version: bool = True)

Bases: NamedTuple

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

classmethod get_configuration(url: str | None, username: str | None, password: str | None, jwtoken: str | None, ssl_verify: bool | str | None, allow_inputs: bool | None = None) → ClientConfig

make_client() → ClientLibrary

allow_http: bool: Alias for field number 5

auto_sync: float: Alias for field number 6

check_version: bool: Alias for field number 11

convergence_wait_max_iter: int: Alias for field number 9

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

events: bool: Alias for field number 7

jwtoken: str | None: Alias for field number 3

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

raise_for_auth_failure: bool: Alias for field number 8

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

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, jwtoken: str | None = None, ssl_verify: bool | str | None = None, 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, client_type: str = None, check_version: bool = True)

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, jwtoken: str | None = None, ssl_verify: bool | str | None = None, 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, client_type: str = None, check_version: bool = True) → None

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 support last 3 minor versions). Raise exception if versions are incompatible, or print warning if the client minor version is lower than the controller minor version. These all are disabled if self.check_version attribute is set to False.

Raises:: InitializationError – If the controller version is incompatible.
Returns:: None

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.
autostart – Parameter for autostart (enabled, priority, delay).
node_staging – Parameter for node staging (enabled, abort_on_failure, start_remaining).

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(*categories: DiagnosticsCategory) → dict

Return selected diagnostics data as a JSON object.

Parameters:: categories – List of diagnostics categories to fetch. DEPRECATED: If not provided, provide DiagnosticsCategory.ALL to fetch all diagnostics data.
Returns:: The diagnostics 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 | None = None, 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 – DEPRECATED: Offline mode has been removed.
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: 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.

VERSION = 2.10.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 module

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 module

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 module

exception virl2_client.exceptions.APIError(message: str, *, request: Request, response: Response): Bases: VirlException, HTTPStatusError

exception virl2_client.exceptions.AnnotationNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.ControllerNotFound

Bases: VirlException

__init__() → None

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

exception virl2_client.exceptions.InvalidTopologySchema: Bases: VirlException

exception virl2_client.exceptions.LabNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.LabRepositoryNotFound: 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: VirlException

exception virl2_client.exceptions.PyatsNotInstalled: Bases: PyatsException

exception virl2_client.exceptions.SmartAnnotationNotFound: Bases: ElementNotFound

exception virl2_client.exceptions.VirlException: Bases: Exception

virl2_client.utils module

class virl2_client.utils.OptInStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ACCEPTED = 'accepted'

DECLINED = 'declined'

UNSET = 'unset'

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 module

class virl2_client.virl2_client.ClientConfig(url: str | None = None, username: str | None = None, password: str | None = None, jwtoken: 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, check_version: bool = True)

Bases: NamedTuple

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

classmethod get_configuration(url: str | None, username: str | None, password: str | None, jwtoken: str | None, ssl_verify: bool | str | None, allow_inputs: bool | None = None) → ClientConfig

make_client() → ClientLibrary

allow_http: bool: Alias for field number 5

auto_sync: float: Alias for field number 6

check_version: bool: Alias for field number 11

convergence_wait_max_iter: int: Alias for field number 9

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

events: bool: Alias for field number 7

jwtoken: str | None: Alias for field number 3

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

raise_for_auth_failure: bool: Alias for field number 8

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

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, jwtoken: str | None = None, ssl_verify: bool | str | None = None, 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, client_type: str = None, check_version: bool = True)

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, jwtoken: str | None = None, ssl_verify: bool | str | None = None, 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, client_type: str = None, check_version: bool = True) → None

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 support last 3 minor versions). Raise exception if versions are incompatible, or print warning if the client minor version is lower than the controller minor version. These all are disabled if self.check_version attribute is set to False.

Raises:: InitializationError – If the controller version is incompatible.
Returns:: None

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.
autostart – Parameter for autostart (enabled, priority, delay).
node_staging – Parameter for node staging (enabled, abort_on_failure, start_remaining).

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(*categories: DiagnosticsCategory) → dict

Return selected diagnostics data as a JSON object.

Parameters:: categories – List of diagnostics categories to fetch. DEPRECATED: If not provided, provide DiagnosticsCategory.ALL to fetch all diagnostics data.
Returns:: The diagnostics 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 | None = None, 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 – DEPRECATED: Offline mode has been removed.
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: 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.

VERSION = 2.10.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.

jwtoken: str | None

password: str | None

url: str

username: str | None

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

class virl2_client.virl2_client.DiagnosticsCategory(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ALL = 'all'

COMPUTES = 'computes'

LABS = 'labs'

LAB_EVENTS = 'lab_events'

LICENSING = 'licensing'

NODE_DEFINITIONS = 'node_definitions'

NODE_LAUNCH_QUEUE = 'node_launch_queue'

SERVICES = 'services'

STARTUP_SCHEDULER = 'startup_scheduler'

USER_LIST = 'user_list'

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

Bases: object

__init__(version_str: str) → None

major_differs(other: Version) → bool

major_lt(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