Arista CloudVision Python Library
The Arista CloudVision Python library provides access to Arista's CloudVision APIs for use in Python applications.
Documentation
API Documentation for this library can be found here.
Documentation for CloudVision's Resource APIs can be found here.
Documentation for generic access to CloudVision can be found at CloudVision Connector.
Installation
Install via pip:
pip install --upgrade cloudvision
Or from source:
python setup.py install
Requirements
- CloudVision Resource APIs: Python 3.7+
- CloudVision Connector: Python 3.7+
- Examples: Python 3.7+
Resource APIs
Cloudvision APIs are state based, resource-oriented APIs modeled in Protobuf and accessed over gRPC using a standardized set of RPC verbs.
CloudVision is a powerful platform that processes and stores tremendous amounts of network data. It knows the topology of the network, device configuration, interface activity and other network events. These APIs allow access to fleet-wide data access and control, forming a management-plane with consistent usage.
For example, consider the following script that gets all existing and then watches for new CloudVision events of critical severity and notifies an administrator when raised and notes it on the event:
import time
import google.protobuf.wrappers_pb2
import grpc
from arista.event.v1 import models, services
# setup credentials as channelCredentials
with grpc.secure_channel("www.arista.io:443", channelCredentials) as channel:
event_stub = services.EventServiceStub(channel)
event_annotation_stub = services.EventAnnotationConfigServiceStub(channel)
event_watch_request = services.EventStreamRequest(
partial_eq_filter=[models.Event(severity=models.EVENT_SEVERITY_CRITICAL)],
)
for resp in event_stub.Subscribe(event_watch_request):
print(f"Critical event {resp.title.value} raised at {resp.key.timestamp}")
# send alert here via email, webhook, or incident reporting tool
# then make a note on the event indicating an alert has been sent
now_ms = int(time.time() * 1000)
notes_to_set = {
now_ms: models.EventNoteConfig(
note=google.protobuf.wrappers_pb2.StringValue(
value="Administrator alerted",
),
),
}
annotation_config = models.EventAnnotationConfig(
key=resp.key,
notes=models.EventNotesConfig(
notes=notes_to_set,
),
)
event_note_update = services.EventAnnotationConfigSetRequest(value=annotation_config)
event_annotation_stub.Set(event_note_update)
Async support
Starting from version 1.26.1+ it is possible to use alternative GRPC client based on
grpclib,that supports pure-python async stubs.
This stubs could be found in cloudvision.api.arista module. More specific information and specifications are provided in
the API documentation
While two types of stubs are supported, it is recommended to use ones from cloudvision.api.arista,
as they use more idiomatic structure, naming conventions and leverage standard python library
to describe data model.
Example below shows how to retrieve a list of devices:
import asyncio
from cloudvision.api.client import AsyncCVClient
from cloudvision.api.arista.inventory.v1 import DeviceServiceStub, DeviceStreamRequest
async def get_devices():
client = AsyncCVClient.from_token('<your service account token>', 'your-cvp.io')
# get channel
with client as channel:
# pass it to the stub
service = DeviceServiceStub(channel)
# execute one of stub's methods
async for item in service.get_all(DeviceStreamRequest()):
print(item)
asyncio.run(get_devices())
CloudVision Connector
CloudVision Connector is a Python implementation of a GRPC client for CloudVision. It takes care of getting and publishing data and datasets, and also provides utilities for data representation.
Getting started
This is a small example advertising a few of the GRPC client capabilities. This example prints info from all devices streaming into CloudVision.
targetDataset = "analytics"
path = ["DatasetInfo", "Devices"]
# No filtering done on keys, accept all
keys = []
ProtoBufQuery = CreateQuery([(path, keys)], targetDataset)
with GRPCClient("my-cv-host:9900") as client:
for notifBatch in client.Get([query]):
for notif in notifBatch["notifications"]:
# Get timestamp for all update here with notif.Timestamp
PrettyPrint(notif["updates"])
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file cloudvision-1.30.5.tar.gz.
File metadata
- Download URL: cloudvision-1.30.5.tar.gz
- Upload date:
- Size: 817.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.2.0 CPython/3.12.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
3c92b46f76d050c8a07ec28fc31e320d32ea5d88220b1e8ad5458ad09df33abc
|
|
| MD5 |
73d34cf1c0901462601e438c99146d27
|
|
| BLAKE2b-256 |
ac926107093ca2aacb294e541530f5fd98c702a6bb87c4ce161186aca5e4c39c
|
File details
Details for the file cloudvision-1.30.5-py3-none-any.whl.
File metadata
- Download URL: cloudvision-1.30.5-py3-none-any.whl
- Upload date:
- Size: 925.5 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.2.0 CPython/3.12.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
ef2d8a15361f6530fa74c69bcb9fa5e7989edb5cdbe45abf05023e3537856637
|
|
| MD5 |
d1604d31667eb1ba1180056f5defe756
|
|
| BLAKE2b-256 |
f777eed543d43dcb9d4dfb278e2564520fa552271c3198e7882b3a7c9ef1b687
|
