# Copyright (c) 2025 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the COPYING file.
# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: arista/configstatus.v1/configstatus.proto, arista/configstatus.v1/services.gen.proto
# plugin: python-aristaproto
# This file has been @generated
__all__ = (
"ErrorCode",
"DiffOp",
"ConfigFilterCode",
"ConfigSyncCode",
"ConfigSourceType",
"ConfigType",
"ConfigError",
"ConfigErrors",
"DiffEntry",
"DiffEntries",
"ConfigSource",
"ConfigSources",
"ConfigSummary",
"ConfigKey",
"Configuration",
"ConfigDiffKey",
"ConfigDiff",
"SummaryKey",
"Summary",
"SecurityProfile",
"SecurityProfileComplianceSummary",
"SecurityProfileDiffSummary",
"SecurityProfileDiff",
"MetaResponse",
"ConfigDiffRequest",
"ConfigDiffResponse",
"ConfigDiffSomeRequest",
"ConfigDiffSomeResponse",
"ConfigDiffStreamRequest",
"ConfigDiffStreamResponse",
"ConfigDiffBatchedStreamRequest",
"ConfigDiffBatchedStreamResponse",
"ConfigurationRequest",
"ConfigurationResponse",
"ConfigurationSomeRequest",
"ConfigurationSomeResponse",
"ConfigurationStreamRequest",
"ConfigurationStreamResponse",
"ConfigurationBatchedStreamRequest",
"ConfigurationBatchedStreamResponse",
"SecurityProfileRequest",
"SecurityProfileResponse",
"SecurityProfileSomeRequest",
"SecurityProfileSomeResponse",
"SecurityProfileStreamRequest",
"SecurityProfileStreamResponse",
"SecurityProfileBatchedStreamRequest",
"SecurityProfileBatchedStreamResponse",
"SecurityProfileDiffRequest",
"SecurityProfileDiffResponse",
"SecurityProfileDiffSomeRequest",
"SecurityProfileDiffSomeResponse",
"SecurityProfileDiffStreamRequest",
"SecurityProfileDiffStreamResponse",
"SecurityProfileDiffBatchedStreamRequest",
"SecurityProfileDiffBatchedStreamResponse",
"SecurityProfileDiffSummaryRequest",
"SecurityProfileDiffSummaryResponse",
"SecurityProfileDiffSummarySomeRequest",
"SecurityProfileDiffSummarySomeResponse",
"SecurityProfileDiffSummaryStreamRequest",
"SecurityProfileDiffSummaryStreamResponse",
"SecurityProfileDiffSummaryBatchedStreamRequest",
"SecurityProfileDiffSummaryBatchedStreamResponse",
"SummaryRequest",
"SummaryResponse",
"SummarySomeRequest",
"SummarySomeResponse",
"SummaryStreamRequest",
"SummaryStreamResponse",
"SummaryBatchedStreamRequest",
"SummaryBatchedStreamResponse",
"ConfigDiffServiceStub",
"ConfigDiffServiceBase",
"ConfigurationServiceStub",
"ConfigurationServiceBase",
"SecurityProfileServiceStub",
"SecurityProfileServiceBase",
"SecurityProfileDiffServiceStub",
"SecurityProfileDiffServiceBase",
"SecurityProfileDiffSummaryServiceStub",
"SecurityProfileDiffSummaryServiceBase",
"SummaryServiceStub",
"SummaryServiceBase",
)
from dataclasses import dataclass
from datetime import datetime
from typing import (
TYPE_CHECKING,
AsyncIterator,
Dict,
List,
Optional,
)
import aristaproto
import grpclib
from aristaproto.grpc.grpclib_server import ServiceBase
if TYPE_CHECKING:
import grpclib.server
from aristaproto.grpc.grpclib_client import MetadataLike
from grpclib.metadata import Deadline
[docs]
class ErrorCode(aristaproto.Enum):
"""
ErrorCode indicates warnings and errors produced during computing config
"""
UNSPECIFIED = 0
"""ERROR_CODE_UNSPECIFIED uninitialized value"""
DEVICE_WARNING = 1
"""ERROR_CODE_DEVICE_WARNING indicates device warning"""
DEVICE_ERROR = 2
"""ERROR_CODE_DEVICE_ERROR indicates device error"""
UNREACHABLE_DEVICE = 3
"""ERROR_CODE_UNREACHABLE_DEVICE indicates the device cannot be reached"""
CONFIG_FILTER_ERROR = 4
"""
ERROR_CODE_CONFIG_FILTER_ERROR indicates error from partial config management filters
"""
INTERNAL = 5
"""ERROR_CODE_INTERNAL indicates internal errors"""
PROPOSED_CONFIG_CHECK = 6
"""
ERROR_CODE_PROPOSED_CONFIG_CHECK indicates an issue with the content of proposed configuration.
"""
[docs]
class DiffOp(aristaproto.Enum):
"""
DiffOp is the operation to a line from one side of diff to get to another
"""
UNSPECIFIED = 0
"""DIFF_OP_UNSPECIFIED uninitialized"""
NOP = 1
"""DIFF_OP_NOP indicates no change. A and B are identical at this line"""
IGNORE = 2
"""
DIFF_OP_IGNORE indicates a line that's ignored in either A or B.
One of a_line_num or b_line_num will be -1
"""
ADD = 3
"""DIFF_OP_ADD is an addition of a line from A"""
DELETE = 4
"""DIFF_OP_DELETE is deletion of a line from B"""
CHANGE = 5
"""DIFF_OP_CHANGE is a modification to a line in A"""
[docs]
class ConfigFilterCode(aristaproto.Enum):
"""ConfigFilterCode indicates if a config line matches PCM filter(s)"""
UNSPECIFIED = 0
"""CONFIG_FILTER_CODE_UNSPECIFIED uninitialized"""
MANAGED_LINE = 1
"""
CONFIG_FILTER_CODE_MANAGED_LINE indicates config line matches managed PCM filter hence is managed
"""
UNMANAGED_LINE = 2
"""
CONFIG_FILTER_CODE_UNMANAGED_LINE indicates config line matches unmanaged PCM filter hence is not managed
"""
[docs]
class ConfigSyncCode(aristaproto.Enum):
"""ConfigSyncCode indicates config synchronization status"""
UNSPECIFIED = 0
"""CONFIG_SYNC_CODE_UNSPECIFIED uninitialized"""
IN_SYNC = 1
"""
CONFIG_SYNC_CODE_IN_SYNC indicates designed config and running config are identical
"""
OUT_OF_SYNC = 2
"""
CONFIG_SYNC_CODE_OUT_OF_SYNC indicates designed config and running config are not identical
"""
[docs]
class ConfigSourceType(aristaproto.Enum):
"""
ConfigSourceType indicates the type of source for the proposed/designed configuration
for the device
"""
UNSPECIFIED = 0
"""CONFIG_SOURCE_TYPE_UNSPECIFIED uninitialized value"""
NETWORK_PROVISIONING_CONFIGLET = 1
"""
CONFIG_SOURCE_TYPE_NETWORK_PROVISIONING_CONFIGLET - configlet created from
the network provisioning workflow.
"""
STUDIO = 2
"""CONFIG_SOURCE_TYPE_STUDIO - config generated from a regular studio"""
STUDIO_STATIC = 3
"""
CONFIG_SOURCE_TYPE_STUDIO_STATIC - static config from studios framework
"""
HIERARCHY = 5
"""CONFIG_SOURCE_TYPE_HIERARCHY - config from the hierarchy frameork"""
[docs]
class ConfigType(aristaproto.Enum):
"""ConfigType - specifies type of config"""
UNSPECIFIED = 0
"""CONFIG_TYPE_UNSPECIFIED uninitialized value"""
RUNNING_CONFIG = 1
"""CONFIG_TYPE_RUNNING_CONFIG uninitialized value"""
DESIGNED_CONFIG = 2
"""CONFIG_TYPE_DESIGNED_CONFIG uninitialized value"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigError(aristaproto.Message):
"""
ConfigError represents errors reported by CVP/EOS when handling device configuration
"""
error_code: "ErrorCode" = aristaproto.enum_field(1)
"""error_code - type of error."""
error_msg: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""error_msg - details of error"""
line_num: Optional[int] = aristaproto.message_field(3, wraps=aristaproto.TYPE_INT32)
"""line_num represents line number, if any"""
configlet_name: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""
configlet_name represents the originating configlet name. Configlet_name
and line_num point to the line where config warning or config error originate.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigErrors(aristaproto.Message):
"""ConfigErrors list of errors"""
values: List["ConfigError"] = aristaproto.message_field(1)
"""values - list of errors"""
[docs]
@dataclass(eq=False, repr=False)
class DiffEntry(aristaproto.Message):
"""DiffEntry represents one entry in a Diff"""
op: "DiffOp" = aristaproto.enum_field(1)
"""op specifies type of diff operation"""
a_line_num: Optional[int] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_INT32
)
"""a_line_num indicates the line number in A where the diff is applied"""
b_line_num: Optional[int] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_INT32
)
"""b_line_num indicates the line number in B where this diff is applied"""
b_parent_line_num: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_INT32
)
"""
b_parent_line_num represents the line number of the parent command in B
"""
a_line: Optional[str] = aristaproto.message_field(5, wraps=aristaproto.TYPE_STRING)
"""a_line is the content of config line in A"""
b_line: Optional[str] = aristaproto.message_field(6, wraps=aristaproto.TYPE_STRING)
"""b_line is the content of config line in B"""
a_filter_code: "ConfigFilterCode" = aristaproto.enum_field(7)
"""a_filter_code represents the config filter code for the line in A"""
b_filter_code: "ConfigFilterCode" = aristaproto.enum_field(8)
"""b_filter_code represents the config filter code for the line in B"""
a_parent_line_num: Optional[int] = aristaproto.message_field(
9, wraps=aristaproto.TYPE_INT32
)
"""
a_parent_line_num represents the line number of the parent command in A
"""
is_a_default_command: Optional[bool] = aristaproto.message_field(
10, wraps=aristaproto.TYPE_BOOL
)
"""
is_a_default_command indicates that config line in A is a default
command that is generated by EOS.
"""
[docs]
@dataclass(eq=False, repr=False)
class DiffEntries(aristaproto.Message):
"""DiffEntries indicates potential multiple lines of config diff"""
values: List["DiffEntry"] = aristaproto.message_field(1)
"""values - list of diff entries"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigSource(aristaproto.Message):
"""
ConfigSource describes an individual source of the proposed/designed config
"""
source_type: "ConfigSourceType" = aristaproto.enum_field(1)
"""source_type - app type of the config snippet"""
source_id: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""
source_id identifier to distinguish between multiple instances of the source type
source_id is :
configlet name for CONFIG_SOURCE_TYPE_NETWORK_PROVISIONING_CONFIGLET
studio id for CONFIG_SOURCE_TYPE_STUDIO
configlet id for CONFIG_SOURCE_TYPE_STUDIO_STATIC
FixtureInstance id for CONFIG_SOURCE_TYPE_HIERARCHY
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigSources(aristaproto.Message):
"""ConfigSources is the list of sources of the proposed/designed config"""
values: List["ConfigSource"] = aristaproto.message_field(1)
"""values - ordered list of all the config snippets present in DC"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigSummary(aristaproto.Message):
"""ConfigSummary represents device configuration summary."""
sync: "ConfigSyncCode" = aristaproto.enum_field(1)
"""sync - state of DC vs RC"""
nop_lines: Optional[int] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_INT32
)
"""nop_lines - Number of lines with code no-operation"""
ignored_lines: Optional[int] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_INT32
)
"""ignored_lines - Number of lines with code IGNORE"""
added_lines: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_INT32
)
"""added_lines - Number of lines with code ADD"""
deleted_lines: Optional[int] = aristaproto.message_field(
5, wraps=aristaproto.TYPE_INT32
)
"""deleted_lines - Number of lines with code DELETE"""
changed_lines: Optional[int] = aristaproto.message_field(
6, wraps=aristaproto.TYPE_INT32
)
"""changed_lines - Number of lines with code CHANGE"""
designed_config_errors: Optional[int] = aristaproto.message_field(
7, wraps=aristaproto.TYPE_INT32
)
"""designed_config_errors - Number of designed config errors"""
designed_config_warnings: Optional[int] = aristaproto.message_field(
8, wraps=aristaproto.TYPE_INT32
)
"""designed_config_warnings - Number of designed config warnings"""
running_config_update_time: datetime = aristaproto.message_field(9)
"""
running_config_update_time - Timestamp at which running config is updated
"""
designed_config_update_time: datetime = aristaproto.message_field(10)
"""
designed_config_update_time - Timestamp at which designed config is updated
"""
running_config_uri: Optional[str] = aristaproto.message_field(
11, wraps=aristaproto.TYPE_STRING
)
"""
running_config_uri - The HTTP URI client can use to GET running config and associated errors
"""
designed_config_uri: Optional[str] = aristaproto.message_field(
12, wraps=aristaproto.TYPE_STRING
)
"""
designed_config_uri - The HTTP URI client can use to GET designed config and associated errors
"""
diff_uri: Optional[str] = aristaproto.message_field(
13, wraps=aristaproto.TYPE_STRING
)
"""
diff_uri - The HTTP URI client can use to GET config diff and associated errors
"""
digest: Optional[str] = aristaproto.message_field(14, wraps=aristaproto.TYPE_STRING)
"""digest (SHA-256) of the config diff."""
[docs]
@dataclass(eq=False, repr=False)
class ConfigKey(aristaproto.Message):
"""ConfigKey uniquely identifies a config request."""
device_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""device_id is the serial number of the device"""
type: "ConfigType" = aristaproto.enum_field(2)
"""type describes the config type"""
[docs]
@dataclass(eq=False, repr=False)
class Configuration(aristaproto.Message):
"""Configuration represents device's CLI configuration"""
key: "ConfigKey" = aristaproto.message_field(1)
"""key specifies the device and type of config"""
uri: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
uri represents the HTTP URI client can use to GET config body and associated errors
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffKey(aristaproto.Message):
"""ConfigDiffKey uniquely identifies a configuration diff request"""
a_device_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""
a_device_id is the serial number of the device on A side (left hand side)
"""
a_type: "ConfigType" = aristaproto.enum_field(2)
"""a_type is the config type on A side (left hand side)"""
a_time: datetime = aristaproto.message_field(3)
"""
a_time is the time at which to fetch config on A side (left hand side)
"""
b_device_id: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""
b_device_id is the serial number of the device on B side (right hand side)
"""
b_type: "ConfigType" = aristaproto.enum_field(5)
"""b_type is the config type on B side (right hand side)"""
b_time: datetime = aristaproto.message_field(6)
"""
b_time is the time at which to fetch config on B side (right hand side)
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiff(aristaproto.Message):
"""ConfigDiff - specifies the diff request"""
key: "ConfigDiffKey" = aristaproto.message_field(1)
"""key represents config diff key"""
uri: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
uri represents the HTTP URI client can use to GET config diff and associated errors
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryKey(aristaproto.Message):
"""SummaryKey uniquely identifies a device summary request"""
device_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""device_id is the serial number of the device"""
[docs]
@dataclass(eq=False, repr=False)
class Summary(aristaproto.Message):
"""Summary - Describes the device's diff summary"""
key: "SummaryKey" = aristaproto.message_field(1)
"""key spefies the device"""
summary: "ConfigSummary" = aristaproto.message_field(2)
"""summary - diff summary of the device"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfile(aristaproto.Message):
"""
SecurityProfile holds the EOS configuration for the security profile.
"""
key: "ConfigKey" = aristaproto.message_field(1)
"""key uniquely identifies the configuration"""
config: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""config describes the security profile config body"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileComplianceSummary(aristaproto.Message):
"""
SecurityProfileComplianceSummary holds the compliance summary for
security profile configuration.
"""
sync: "ConfigSyncCode" = aristaproto.enum_field(1)
"""sync - sync state of the device"""
nop_lines: Optional[int] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_INT32
)
"""nop_lines is the number of lines with code no-operation"""
ignored_lines: Optional[int] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_INT32
)
"""ignored_lines is the number of lines with code IGNORE"""
added_lines: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_INT32
)
"""added_lines is the number of lines with code ADD"""
deleted_lines: Optional[int] = aristaproto.message_field(
5, wraps=aristaproto.TYPE_INT32
)
"""deleted_lines is the number of lines with code DELETE"""
changed_lines: Optional[int] = aristaproto.message_field(
6, wraps=aristaproto.TYPE_INT32
)
"""changed_lines is the Number of lines with code CHANGE"""
digest: Optional[str] = aristaproto.message_field(7, wraps=aristaproto.TYPE_STRING)
"""
digest is the digest (SHA-256 hash) of the security profile configuration diff
"""
running_config_update_time: datetime = aristaproto.message_field(8)
"""
running_config_update_time is the timestamp at which running security profile
or running config is updated
"""
designed_config_update_time: datetime = aristaproto.message_field(9)
"""
designed_config_update_time is the timestamp at which designed security profile or
designed config is updated
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummary(aristaproto.Message):
"""
SecurityProfileDiffSummary holds device compliance summary w.r.t security profile.
"""
key: "SummaryKey" = aristaproto.message_field(1)
"""key uniquely identifies the Summary"""
summary: "SecurityProfileComplianceSummary" = aristaproto.message_field(2)
"""summary is the security profile configuration compliance summary"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiff(aristaproto.Message):
"""SecurityProfileDiff holds the security profile configuration diff."""
key: "ConfigDiffKey" = aristaproto.message_field(1)
"""key represents security profile config diff key"""
diff: "DiffEntries" = aristaproto.message_field(2)
"""diff represents diff entries"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffRequest(aristaproto.Message):
""" """
key: "ConfigDiffKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a ConfigDiff instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffResponse(aristaproto.Message):
""" """
value: "ConfigDiff" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
ConfigDiff instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigDiffKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffSomeResponse(aristaproto.Message):
""" """
value: "ConfigDiff" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
ConfigDiff instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigDiff"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each ConfigDiff at end.
* Each ConfigDiff response is fully-specified (all fields set).
* start: Returns the state of each ConfigDiff at start, followed by updates until now.
* Each ConfigDiff response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigDiff at start, followed by updates
until end.
* Each ConfigDiff response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffStreamResponse(aristaproto.Message):
""" """
value: "ConfigDiff" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""Time holds the timestamp of this ConfigDiff's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the ConfigDiff value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigDiff"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each ConfigDiff at end.
* Each ConfigDiff response is fully-specified (all fields set).
* start: Returns the state of each ConfigDiff at start, followed by updates until now.
* Each ConfigDiff response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigDiff at start, followed by updates
until end.
* Each ConfigDiff response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigDiffBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigDiffStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationRequest(aristaproto.Message):
""" """
key: "ConfigKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a Configuration instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationResponse(aristaproto.Message):
""" """
value: "Configuration" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
Configuration instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationSomeResponse(aristaproto.Message):
""" """
value: "Configuration" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
Configuration instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Configuration"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each Configuration at end.
* Each Configuration response is fully-specified (all fields set).
* start: Returns the state of each Configuration at start, followed by updates until now.
* Each Configuration response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Configuration at start, followed by updates
until end.
* Each Configuration response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationStreamResponse(aristaproto.Message):
""" """
value: "Configuration" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""Time holds the timestamp of this Configuration's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the Configuration value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Configuration"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each Configuration at end.
* Each Configuration response is fully-specified (all fields set).
* start: Returns the state of each Configuration at start, followed by updates until now.
* Each Configuration response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Configuration at start, followed by updates
until end.
* Each Configuration response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigurationBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigurationStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileRequest(aristaproto.Message):
""" """
key: "ConfigKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a SecurityProfile instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileResponse(aristaproto.Message):
""" """
value: "SecurityProfile" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfile instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileSomeResponse(aristaproto.Message):
""" """
value: "SecurityProfile" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfile instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfile"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfile at end.
* Each SecurityProfile response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfile at start, followed by updates until now.
* Each SecurityProfile response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfile at start, followed by updates
until end.
* Each SecurityProfile response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileStreamResponse(aristaproto.Message):
""" """
value: "SecurityProfile" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""
Time holds the timestamp of this SecurityProfile's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the SecurityProfile value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfile"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfile at end.
* Each SecurityProfile response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfile at start, followed by updates until now.
* Each SecurityProfile response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfile at start, followed by updates
until end.
* Each SecurityProfile response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["SecurityProfileStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffRequest(aristaproto.Message):
""" """
key: "ConfigDiffKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a SecurityProfileDiff instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiff" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfileDiff instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigDiffKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSomeResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiff" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfileDiff instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfileDiff"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfileDiff at end.
* Each SecurityProfileDiff response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfileDiff at start, followed by updates until now.
* Each SecurityProfileDiff response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfileDiff at start, followed by updates
until end.
* Each SecurityProfileDiff response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffStreamResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiff" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""
Time holds the timestamp of this SecurityProfileDiff's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the SecurityProfileDiff value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfileDiff"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfileDiff at end.
* Each SecurityProfileDiff response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfileDiff at start, followed by updates until now.
* Each SecurityProfileDiff response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfileDiff at start, followed by updates
until end.
* Each SecurityProfileDiff response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["SecurityProfileDiffStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryRequest(aristaproto.Message):
""" """
key: "SummaryKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a SecurityProfileDiffSummary instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiffSummary" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfileDiffSummary instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummarySomeRequest(aristaproto.Message):
""" """
keys: List["SummaryKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummarySomeResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiffSummary" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
SecurityProfileDiffSummary instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfileDiffSummary"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfileDiffSummary at end.
* Each SecurityProfileDiffSummary response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfileDiffSummary at start, followed by updates until now.
* Each SecurityProfileDiffSummary response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfileDiffSummary at start, followed by updates
until end.
* Each SecurityProfileDiffSummary response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryStreamResponse(aristaproto.Message):
""" """
value: "SecurityProfileDiffSummary" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""
Time holds the timestamp of this SecurityProfileDiffSummary's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the SecurityProfileDiffSummary value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["SecurityProfileDiffSummary"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each SecurityProfileDiffSummary at end.
* Each SecurityProfileDiffSummary response is fully-specified (all fields set).
* start: Returns the state of each SecurityProfileDiffSummary at start, followed by updates until now.
* Each SecurityProfileDiffSummary response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each SecurityProfileDiffSummary at start, followed by updates
until end.
* Each SecurityProfileDiffSummary response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class SecurityProfileDiffSummaryBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["SecurityProfileDiffSummaryStreamResponse"] = (
aristaproto.message_field(1)
)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryRequest(aristaproto.Message):
""" """
key: "SummaryKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a Summary instance to retrieve.
This value must be populated.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryResponse(aristaproto.Message):
""" """
value: "Summary" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time carries the (UTC) timestamp of the last-modification of the
Summary instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummarySomeRequest(aristaproto.Message):
""" """
keys: List["SummaryKey"] = aristaproto.message_field(1)
"""
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the time for which you are interested in the data.
If no time is given, the server will use the time at which it makes the request.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummarySomeResponse(aristaproto.Message):
""" """
value: "Summary" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
Summary instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Summary"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each Summary at end.
* Each Summary response is fully-specified (all fields set).
* start: Returns the state of each Summary at start, followed by updates until now.
* Each Summary response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Summary at start, followed by updates
until end.
* Each Summary response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryStreamResponse(aristaproto.Message):
""" """
value: "Summary" = aristaproto.message_field(1)
"""
Value is a value deemed relevant to the initiating request.
This structure will always have its key-field populated. Which other fields are
populated, and why, depends on the value of Operation and what triggered this notification.
"""
time: datetime = aristaproto.message_field(2)
"""Time holds the timestamp of this Summary's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the Summary value in this response should be considered.
Under non-subscribe requests, this value should always be INITIAL. In a subscription,
once all initial data is streamed and the client begins to receive modification updates,
you should not see INITIAL again.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Summary"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a GetAll/Subscribe.
This requires all provided fields to be equal to the response.
While transparent to users, this field also allows services to optimize internal
subscriptions if filter(s) are sufficiently specific.
"""
time: "__time__.TimeBounds" = aristaproto.message_field(3)
"""
TimeRange allows limiting response data to within a specified time window.
If this field is populated, at least one of the two time fields are required.
For GetAll, the fields start and end can be used as follows:
* end: Returns the state of each Summary at end.
* Each Summary response is fully-specified (all fields set).
* start: Returns the state of each Summary at start, followed by updates until now.
* Each Summary response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Summary at start, followed by updates
until end.
* Each Summary response at start is fully-specified, but updates until end may
be partial.
"""
max_messages: Optional[int] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_UINT32
)
"""
MaxMessages limits the maximum number of messages that can be contained in one batch.
MaxMessages is required to be at least 1.
The maximum number of messages in a batch is min(max_messages, INTERNAL_BATCH_LIMIT)
INTERNAL_BATCH_LIMIT is set based on the maximum message size.
"""
[docs]
@dataclass(eq=False, repr=False)
class SummaryBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["SummaryStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
class ConfigDiffServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
config_diff_request: "ConfigDiffRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigDiffResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.ConfigDiffService/GetOne",
config_diff_request,
ConfigDiffResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
config_diff_some_request: "ConfigDiffSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigDiffSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigDiffService/GetSome",
config_diff_some_request,
ConfigDiffSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
config_diff_stream_request: "ConfigDiffStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigDiffStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigDiffService/GetAll",
config_diff_stream_request,
ConfigDiffStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
config_diff_stream_request: "ConfigDiffStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigDiffStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigDiffService/Subscribe",
config_diff_stream_request,
ConfigDiffStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
config_diff_batched_stream_request: "ConfigDiffBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigDiffBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigDiffService/GetAllBatched",
config_diff_batched_stream_request,
ConfigDiffBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
config_diff_batched_stream_request: "ConfigDiffBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigDiffBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigDiffService/SubscribeBatched",
config_diff_batched_stream_request,
ConfigDiffBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class ConfigurationServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
configuration_request: "ConfigurationRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigurationResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.ConfigurationService/GetOne",
configuration_request,
ConfigurationResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
configuration_some_request: "ConfigurationSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigurationSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigurationService/GetSome",
configuration_some_request,
ConfigurationSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
configuration_stream_request: "ConfigurationStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigurationStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigurationService/GetAll",
configuration_stream_request,
ConfigurationStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
configuration_stream_request: "ConfigurationStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigurationStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigurationService/Subscribe",
configuration_stream_request,
ConfigurationStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
configuration_batched_stream_request: "ConfigurationBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigurationBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigurationService/GetAllBatched",
configuration_batched_stream_request,
ConfigurationBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
configuration_batched_stream_request: "ConfigurationBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigurationBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.ConfigurationService/SubscribeBatched",
configuration_batched_stream_request,
ConfigurationBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class SecurityProfileServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
security_profile_request: "SecurityProfileRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "SecurityProfileResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.SecurityProfileService/GetOne",
security_profile_request,
SecurityProfileResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
security_profile_some_request: "SecurityProfileSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileService/GetSome",
security_profile_some_request,
SecurityProfileSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
security_profile_stream_request: "SecurityProfileStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileService/GetAll",
security_profile_stream_request,
SecurityProfileStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
security_profile_stream_request: "SecurityProfileStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileService/Subscribe",
security_profile_stream_request,
SecurityProfileStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
security_profile_batched_stream_request: "SecurityProfileBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileService/GetAllBatched",
security_profile_batched_stream_request,
SecurityProfileBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
security_profile_batched_stream_request: "SecurityProfileBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileService/SubscribeBatched",
security_profile_batched_stream_request,
SecurityProfileBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class SecurityProfileDiffServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
security_profile_diff_request: "SecurityProfileDiffRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "SecurityProfileDiffResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.SecurityProfileDiffService/GetOne",
security_profile_diff_request,
SecurityProfileDiffResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
security_profile_diff_some_request: "SecurityProfileDiffSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffService/GetSome",
security_profile_diff_some_request,
SecurityProfileDiffSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
security_profile_diff_stream_request: "SecurityProfileDiffStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffService/GetAll",
security_profile_diff_stream_request,
SecurityProfileDiffStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
security_profile_diff_stream_request: "SecurityProfileDiffStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffService/Subscribe",
security_profile_diff_stream_request,
SecurityProfileDiffStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
security_profile_diff_batched_stream_request: "SecurityProfileDiffBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffService/GetAllBatched",
security_profile_diff_batched_stream_request,
SecurityProfileDiffBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
security_profile_diff_batched_stream_request: "SecurityProfileDiffBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffService/SubscribeBatched",
security_profile_diff_batched_stream_request,
SecurityProfileDiffBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class SecurityProfileDiffSummaryServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
security_profile_diff_summary_request: "SecurityProfileDiffSummaryRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "SecurityProfileDiffSummaryResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetOne",
security_profile_diff_summary_request,
SecurityProfileDiffSummaryResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
security_profile_diff_summary_some_request: "SecurityProfileDiffSummarySomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSummarySomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetSome",
security_profile_diff_summary_some_request,
SecurityProfileDiffSummarySomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
security_profile_diff_summary_stream_request: "SecurityProfileDiffSummaryStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSummaryStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetAll",
security_profile_diff_summary_stream_request,
SecurityProfileDiffSummaryStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
security_profile_diff_summary_stream_request: "SecurityProfileDiffSummaryStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSummaryStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/Subscribe",
security_profile_diff_summary_stream_request,
SecurityProfileDiffSummaryStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
security_profile_diff_summary_batched_stream_request: "SecurityProfileDiffSummaryBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSummaryBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetAllBatched",
security_profile_diff_summary_batched_stream_request,
SecurityProfileDiffSummaryBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
security_profile_diff_summary_batched_stream_request: "SecurityProfileDiffSummaryBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SecurityProfileDiffSummaryBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/SubscribeBatched",
security_profile_diff_summary_batched_stream_request,
SecurityProfileDiffSummaryBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class SummaryServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
summary_request: "SummaryRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "SummaryResponse":
""" """
return await self._unary_unary(
"/arista.configstatus.v1.SummaryService/GetOne",
summary_request,
SummaryResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
summary_some_request: "SummarySomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SummarySomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SummaryService/GetSome",
summary_some_request,
SummarySomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
summary_stream_request: "SummaryStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SummaryStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SummaryService/GetAll",
summary_stream_request,
SummaryStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
summary_stream_request: "SummaryStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SummaryStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SummaryService/Subscribe",
summary_stream_request,
SummaryStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
summary_batched_stream_request: "SummaryBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SummaryBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SummaryService/GetAllBatched",
summary_batched_stream_request,
SummaryBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
summary_batched_stream_request: "SummaryBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[SummaryBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configstatus.v1.SummaryService/SubscribeBatched",
summary_batched_stream_request,
SummaryBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
from ... import subscriptions as __subscriptions__
from ... import time as __time__
[docs]
class ConfigDiffServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, config_diff_request: "ConfigDiffRequest"
) -> "ConfigDiffResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, config_diff_some_request: "ConfigDiffSomeRequest"
) -> AsyncIterator[ConfigDiffSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, config_diff_stream_request: "ConfigDiffStreamRequest"
) -> AsyncIterator[ConfigDiffStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, config_diff_stream_request: "ConfigDiffStreamRequest"
) -> AsyncIterator[ConfigDiffStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self, config_diff_batched_stream_request: "ConfigDiffBatchedStreamRequest"
) -> AsyncIterator[ConfigDiffBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self, config_diff_batched_stream_request: "ConfigDiffBatchedStreamRequest"
) -> AsyncIterator[ConfigDiffBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[ConfigDiffRequest, ConfigDiffResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self,
stream: "grpclib.server.Stream[ConfigDiffSomeRequest, ConfigDiffSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[ConfigDiffStreamRequest, ConfigDiffStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[ConfigDiffStreamRequest, ConfigDiffStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self, stream: "grpclib.server.Stream[ConfigDiffStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self, stream: "grpclib.server.Stream[ConfigDiffStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[ConfigDiffBatchedStreamRequest, ConfigDiffBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[ConfigDiffBatchedStreamRequest, ConfigDiffBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.ConfigDiffService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigDiffRequest,
ConfigDiffResponse,
),
"/arista.configstatus.v1.ConfigDiffService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffSomeRequest,
ConfigDiffSomeResponse,
),
"/arista.configstatus.v1.ConfigDiffService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffStreamRequest,
ConfigDiffStreamResponse,
),
"/arista.configstatus.v1.ConfigDiffService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffStreamRequest,
ConfigDiffStreamResponse,
),
"/arista.configstatus.v1.ConfigDiffService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigDiffStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.ConfigDiffService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.ConfigDiffService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffBatchedStreamRequest,
ConfigDiffBatchedStreamResponse,
),
"/arista.configstatus.v1.ConfigDiffService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigDiffBatchedStreamRequest,
ConfigDiffBatchedStreamResponse,
),
}
[docs]
class ConfigurationServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, configuration_request: "ConfigurationRequest"
) -> "ConfigurationResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, configuration_some_request: "ConfigurationSomeRequest"
) -> AsyncIterator[ConfigurationSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, configuration_stream_request: "ConfigurationStreamRequest"
) -> AsyncIterator[ConfigurationStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, configuration_stream_request: "ConfigurationStreamRequest"
) -> AsyncIterator[ConfigurationStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self, configuration_batched_stream_request: "ConfigurationBatchedStreamRequest"
) -> AsyncIterator[ConfigurationBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self, configuration_batched_stream_request: "ConfigurationBatchedStreamRequest"
) -> AsyncIterator[ConfigurationBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[ConfigurationRequest, ConfigurationResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self,
stream: "grpclib.server.Stream[ConfigurationSomeRequest, ConfigurationSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[ConfigurationStreamRequest, ConfigurationStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[ConfigurationStreamRequest, ConfigurationStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self, stream: "grpclib.server.Stream[ConfigurationStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self, stream: "grpclib.server.Stream[ConfigurationStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[ConfigurationBatchedStreamRequest, ConfigurationBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[ConfigurationBatchedStreamRequest, ConfigurationBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.ConfigurationService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigurationRequest,
ConfigurationResponse,
),
"/arista.configstatus.v1.ConfigurationService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationSomeRequest,
ConfigurationSomeResponse,
),
"/arista.configstatus.v1.ConfigurationService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationStreamRequest,
ConfigurationStreamResponse,
),
"/arista.configstatus.v1.ConfigurationService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationStreamRequest,
ConfigurationStreamResponse,
),
"/arista.configstatus.v1.ConfigurationService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigurationStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.ConfigurationService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.ConfigurationService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationBatchedStreamRequest,
ConfigurationBatchedStreamResponse,
),
"/arista.configstatus.v1.ConfigurationService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigurationBatchedStreamRequest,
ConfigurationBatchedStreamResponse,
),
}
[docs]
class SecurityProfileServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, security_profile_request: "SecurityProfileRequest"
) -> "SecurityProfileResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, security_profile_some_request: "SecurityProfileSomeRequest"
) -> AsyncIterator[SecurityProfileSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, security_profile_stream_request: "SecurityProfileStreamRequest"
) -> AsyncIterator[SecurityProfileStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, security_profile_stream_request: "SecurityProfileStreamRequest"
) -> AsyncIterator[SecurityProfileStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
security_profile_batched_stream_request: "SecurityProfileBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
security_profile_batched_stream_request: "SecurityProfileBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[SecurityProfileRequest, SecurityProfileResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self,
stream: "grpclib.server.Stream[SecurityProfileSomeRequest, SecurityProfileSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[SecurityProfileStreamRequest, SecurityProfileStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[SecurityProfileStreamRequest, SecurityProfileStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileBatchedStreamRequest, SecurityProfileBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileBatchedStreamRequest, SecurityProfileBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.SecurityProfileService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileRequest,
SecurityProfileResponse,
),
"/arista.configstatus.v1.SecurityProfileService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileSomeRequest,
SecurityProfileSomeResponse,
),
"/arista.configstatus.v1.SecurityProfileService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileStreamRequest,
SecurityProfileStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileStreamRequest,
SecurityProfileStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileBatchedStreamRequest,
SecurityProfileBatchedStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileBatchedStreamRequest,
SecurityProfileBatchedStreamResponse,
),
}
[docs]
class SecurityProfileDiffServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, security_profile_diff_request: "SecurityProfileDiffRequest"
) -> "SecurityProfileDiffResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, security_profile_diff_some_request: "SecurityProfileDiffSomeRequest"
) -> AsyncIterator[SecurityProfileDiffSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, security_profile_diff_stream_request: "SecurityProfileDiffStreamRequest"
) -> AsyncIterator[SecurityProfileDiffStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, security_profile_diff_stream_request: "SecurityProfileDiffStreamRequest"
) -> AsyncIterator[SecurityProfileDiffStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
security_profile_diff_batched_stream_request: "SecurityProfileDiffBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileDiffBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
security_profile_diff_batched_stream_request: "SecurityProfileDiffBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileDiffBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffRequest, SecurityProfileDiffResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSomeRequest, SecurityProfileDiffSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffStreamRequest, SecurityProfileDiffStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffStreamRequest, SecurityProfileDiffStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffBatchedStreamRequest, SecurityProfileDiffBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffBatchedStreamRequest, SecurityProfileDiffBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.SecurityProfileDiffService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileDiffRequest,
SecurityProfileDiffResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSomeRequest,
SecurityProfileDiffSomeResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffStreamRequest,
SecurityProfileDiffStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffStreamRequest,
SecurityProfileDiffStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileDiffStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffBatchedStreamRequest,
SecurityProfileDiffBatchedStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffBatchedStreamRequest,
SecurityProfileDiffBatchedStreamResponse,
),
}
[docs]
class SecurityProfileDiffSummaryServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, security_profile_diff_summary_request: "SecurityProfileDiffSummaryRequest"
) -> "SecurityProfileDiffSummaryResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self,
security_profile_diff_summary_some_request: "SecurityProfileDiffSummarySomeRequest",
) -> AsyncIterator[SecurityProfileDiffSummarySomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self,
security_profile_diff_summary_stream_request: "SecurityProfileDiffSummaryStreamRequest",
) -> AsyncIterator[SecurityProfileDiffSummaryStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self,
security_profile_diff_summary_stream_request: "SecurityProfileDiffSummaryStreamRequest",
) -> AsyncIterator[SecurityProfileDiffSummaryStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
security_profile_diff_summary_batched_stream_request: "SecurityProfileDiffSummaryBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileDiffSummaryBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
security_profile_diff_summary_batched_stream_request: "SecurityProfileDiffSummaryBatchedStreamRequest",
) -> AsyncIterator[SecurityProfileDiffSummaryBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryRequest, SecurityProfileDiffSummaryResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummarySomeRequest, SecurityProfileDiffSummarySomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryStreamRequest, SecurityProfileDiffSummaryStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryStreamRequest, SecurityProfileDiffSummaryStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryBatchedStreamRequest, SecurityProfileDiffSummaryBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[SecurityProfileDiffSummaryBatchedStreamRequest, SecurityProfileDiffSummaryBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileDiffSummaryRequest,
SecurityProfileDiffSummaryResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummarySomeRequest,
SecurityProfileDiffSummarySomeResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummaryStreamRequest,
SecurityProfileDiffSummaryStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummaryStreamRequest,
SecurityProfileDiffSummaryStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
SecurityProfileDiffSummaryStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummaryStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummaryBatchedStreamRequest,
SecurityProfileDiffSummaryBatchedStreamResponse,
),
"/arista.configstatus.v1.SecurityProfileDiffSummaryService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SecurityProfileDiffSummaryBatchedStreamRequest,
SecurityProfileDiffSummaryBatchedStreamResponse,
),
}
[docs]
class SummaryServiceBase(ServiceBase):
""" """
[docs]
async def get_one(self, summary_request: "SummaryRequest") -> "SummaryResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, summary_some_request: "SummarySomeRequest"
) -> AsyncIterator[SummarySomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, summary_stream_request: "SummaryStreamRequest"
) -> AsyncIterator[SummaryStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, summary_stream_request: "SummaryStreamRequest"
) -> AsyncIterator[SummaryStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self, summary_batched_stream_request: "SummaryBatchedStreamRequest"
) -> AsyncIterator[SummaryBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self, summary_batched_stream_request: "SummaryBatchedStreamRequest"
) -> AsyncIterator[SummaryBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[SummaryRequest, SummaryResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_one(request)
await stream.send_message(response)
async def __rpc_get_some(
self, stream: "grpclib.server.Stream[SummarySomeRequest, SummarySomeResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_some,
stream,
request,
)
async def __rpc_get_all(
self,
stream: "grpclib.server.Stream[SummaryStreamRequest, SummaryStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all,
stream,
request,
)
async def __rpc_subscribe(
self,
stream: "grpclib.server.Stream[SummaryStreamRequest, SummaryStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe,
stream,
request,
)
async def __rpc_get_meta(
self, stream: "grpclib.server.Stream[SummaryStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
response = await self.get_meta(request)
await stream.send_message(response)
async def __rpc_subscribe_meta(
self, stream: "grpclib.server.Stream[SummaryStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[SummaryBatchedStreamRequest, SummaryBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.get_all_batched,
stream,
request,
)
async def __rpc_subscribe_batched(
self,
stream: "grpclib.server.Stream[SummaryBatchedStreamRequest, SummaryBatchedStreamResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_batched,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.configstatus.v1.SummaryService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
SummaryRequest,
SummaryResponse,
),
"/arista.configstatus.v1.SummaryService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
SummarySomeRequest,
SummarySomeResponse,
),
"/arista.configstatus.v1.SummaryService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
SummaryStreamRequest,
SummaryStreamResponse,
),
"/arista.configstatus.v1.SummaryService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
SummaryStreamRequest,
SummaryStreamResponse,
),
"/arista.configstatus.v1.SummaryService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
SummaryStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SummaryService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
SummaryStreamRequest,
MetaResponse,
),
"/arista.configstatus.v1.SummaryService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SummaryBatchedStreamRequest,
SummaryBatchedStreamResponse,
),
"/arista.configstatus.v1.SummaryService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
SummaryBatchedStreamRequest,
SummaryBatchedStreamResponse,
),
}