# 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/configlet.v1/configlet.proto, arista/configlet.v1/services.gen.proto
# plugin: python-aristaproto
# This file has been @generated
__all__ = (
"MatchPolicy",
"ConfigletKey",
"Filter",
"Configlet",
"ConfigletConfig",
"ConfigletAssignmentKey",
"ConfigletAssignmentConfig",
"ConfigletAssignment",
"MetaResponse",
"ConfigletRequest",
"ConfigletResponse",
"ConfigletSomeRequest",
"ConfigletSomeResponse",
"ConfigletStreamRequest",
"ConfigletStreamResponse",
"ConfigletBatchedStreamRequest",
"ConfigletBatchedStreamResponse",
"ConfigletAssignmentRequest",
"ConfigletAssignmentResponse",
"ConfigletAssignmentSomeRequest",
"ConfigletAssignmentSomeResponse",
"ConfigletAssignmentStreamRequest",
"ConfigletAssignmentStreamResponse",
"ConfigletAssignmentBatchedStreamRequest",
"ConfigletAssignmentBatchedStreamResponse",
"ConfigletAssignmentConfigRequest",
"ConfigletAssignmentConfigResponse",
"ConfigletAssignmentConfigSomeRequest",
"ConfigletAssignmentConfigSomeResponse",
"ConfigletAssignmentConfigStreamRequest",
"ConfigletAssignmentConfigStreamResponse",
"ConfigletAssignmentConfigBatchedStreamRequest",
"ConfigletAssignmentConfigBatchedStreamResponse",
"ConfigletAssignmentConfigSetRequest",
"ConfigletAssignmentConfigSetResponse",
"ConfigletAssignmentConfigSetSomeRequest",
"ConfigletAssignmentConfigSetSomeResponse",
"ConfigletAssignmentConfigDeleteRequest",
"ConfigletAssignmentConfigDeleteResponse",
"ConfigletAssignmentConfigDeleteSomeRequest",
"ConfigletAssignmentConfigDeleteSomeResponse",
"ConfigletAssignmentConfigDeleteAllRequest",
"ConfigletAssignmentConfigDeleteAllResponse",
"ConfigletConfigRequest",
"ConfigletConfigResponse",
"ConfigletConfigSomeRequest",
"ConfigletConfigSomeResponse",
"ConfigletConfigStreamRequest",
"ConfigletConfigStreamResponse",
"ConfigletConfigBatchedStreamRequest",
"ConfigletConfigBatchedStreamResponse",
"ConfigletConfigSetRequest",
"ConfigletConfigSetResponse",
"ConfigletConfigSetSomeRequest",
"ConfigletConfigSetSomeResponse",
"ConfigletConfigDeleteRequest",
"ConfigletConfigDeleteResponse",
"ConfigletConfigDeleteSomeRequest",
"ConfigletConfigDeleteSomeResponse",
"ConfigletConfigDeleteAllRequest",
"ConfigletConfigDeleteAllResponse",
"ConfigletServiceStub",
"ConfigletServiceBase",
"ConfigletAssignmentServiceStub",
"ConfigletAssignmentServiceBase",
"ConfigletAssignmentConfigServiceStub",
"ConfigletAssignmentConfigServiceBase",
"ConfigletConfigServiceStub",
"ConfigletConfigServiceBase",
)
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 MatchPolicy(aristaproto.Enum):
"""
MatchPolicy specifies how the tag query of a configlet assignment
should be used to resolve devices for its children
"""
UNSPECIFIED = 0
"""
"""
MATCH_FIRST = 1
"""
MATCH_POLICY_MATCH_FIRST dictates that matching devices are used only for the first match amongst its children
"""
MATCH_ALL = 2
"""
MATCH_POLICY_MATCH_ALL dictates that matching devices are used across all children
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletKey(aristaproto.Message):
"""ConfigletKey uniquely identifies a static configlet."""
workspace_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""
workspace_id identifies the workspace within which the static configlet resides
empty string (\"\") stands for the \"mainline\".
"""
configlet_id: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""configlet_id is the static configlet ID."""
[docs]
@dataclass(eq=False, repr=False)
class Filter(aristaproto.Message):
"""Filter is used to filter static configlets."""
include_body: Optional[bool] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_BOOL
)
"""include_body specifies the static configlet body is to be included."""
[docs]
@dataclass(eq=False, repr=False)
class Configlet(aristaproto.Message):
"""
Configlet is the state of a static configlet in a workspace or mainline.
Subscribe and GetAll do not return the \"body\"
Use GetOne to get the body of individual configlets
"""
key: "ConfigletKey" = aristaproto.message_field(1)
"""
"""
display_name: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""display_name is the display name of the static configlet."""
description: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""description is the description of the static configlet."""
migrated_from: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""
migrated_from is populated with the source configlet name when migrated
from network provisioning to studio.
"""
body: Optional[str] = aristaproto.message_field(5, wraps=aristaproto.TYPE_STRING)
"""body is the static configlet body."""
created_at: datetime = aristaproto.message_field(6)
"""created_at is the time when the Configlet was created."""
created_by: Optional[str] = aristaproto.message_field(
7, wraps=aristaproto.TYPE_STRING
)
"""created_by is the user who created the Configlet."""
last_modified_at: datetime = aristaproto.message_field(8)
"""last_modified_at is the time when the Configlet was last modified."""
last_modified_by: Optional[str] = aristaproto.message_field(
9, wraps=aristaproto.TYPE_STRING
)
"""last_modified_by is the user who last modified the Configlet."""
digest: Optional[str] = aristaproto.message_field(10, wraps=aristaproto.TYPE_STRING)
"""
digest is the sha256 hash of the configlet body encoded in hexadecimal.
"""
size: Optional[int] = aristaproto.message_field(11, wraps=aristaproto.TYPE_INT64)
"""size of configlet body in bytes."""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfig(aristaproto.Message):
"""ConfigletConfig updates a static configlet in a workspace."""
key: "ConfigletKey" = aristaproto.message_field(1)
"""
"""
remove: Optional[bool] = aristaproto.message_field(2, wraps=aristaproto.TYPE_BOOL)
"""
remove specifies the static configlet is to be removed from the workspace.
Other data fields are not allowed when this field is set to true.
"""
display_name: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""display_name is the display name of the static configlet."""
description: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""description is the description of the static configlet."""
migrated_from: Optional[str] = aristaproto.message_field(
5, wraps=aristaproto.TYPE_STRING
)
"""
migrated_from is populated with the source configlet name when migrated
from network provisioning to studio.
"""
body: Optional[str] = aristaproto.message_field(6, wraps=aristaproto.TYPE_STRING)
"""body is the static configlet body."""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentKey(aristaproto.Message):
"""ConfigletAssignmentKey uniquely identifies a configlet assignment"""
workspace_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""workspace_id is the unique identifier of the workspace."""
configlet_assignment_id: Optional[str] = aristaproto.message_field(
2, wraps=aristaproto.TYPE_STRING
)
"""
configlet_assignment_id is the unique identifier of the configlet_assignment.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfig(aristaproto.Message):
"""
ConfigletAssignmentConfig are the the inputs to the static configlet studio.
Each assignment assigns a list of configlets to the devices matching the tag query.
Individual assignments can have a list of \"child\" assignments.
The totality of these assignments form a list of tree hierarchies.
Using the corresponding GUI workflow should help explain this structure.
When traversing a tree and assigning static configlets to devices, the following rules
are applied:
1) Tag queries at each level need to resolve to a subset of its parent's queries.
Devices resolved in child assignments but which do not resolve in that of its parent's
are skipped.
2) Match policy determines how the assignment's devices get divied up amongst its children.
"""
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
"""
display_name: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""
"""
description: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""
"""
configlet_ids: "___fmp__.RepeatedString" = aristaproto.message_field(5)
"""configlet_ids is the list of configlets to be assigned"""
query: Optional[str] = aristaproto.message_field(6, wraps=aristaproto.TYPE_STRING)
"""
query is a tag query string that conforms to the CloudVision
tag query language. E.g., the query, `\"datacenter:NYC,SFO AND
sflow:enabled\"`, matches all devices with sflow enabled in
data centers NYC and SFO.
malformed queries result in an error
tags not matching devices are ignored
"""
remove: Optional[bool] = aristaproto.message_field(7, wraps=aristaproto.TYPE_BOOL)
"""
remove indicates whether to remove (`true`) or add (`false`,
unset) the tag assignments involving the studio identified
by the key if the encompassing workspace merges. Other data
fields are not allowed if this field is set to true.
"""
match_policy: "MatchPolicy" = aristaproto.enum_field(8)
"""match_policy is the discriminator for the query field"""
child_assignment_ids: "___fmp__.RepeatedString" = aristaproto.message_field(9)
"""list of child assignments"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignment(aristaproto.Message):
"""
ConfigletAssignment is the state of this assignment in a workspace/mainline
"""
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
"""
display_name: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""
"""
description: Optional[str] = aristaproto.message_field(
4, wraps=aristaproto.TYPE_STRING
)
"""
"""
configlet_ids: "___fmp__.RepeatedString" = aristaproto.message_field(5)
"""configlet_ids is the list of configlets which are assigned"""
query: Optional[str] = aristaproto.message_field(6, wraps=aristaproto.TYPE_STRING)
"""query represents the tag query assigned"""
match_policy: "MatchPolicy" = aristaproto.enum_field(7)
"""match_policy is the discriminator for the query field"""
child_assignment_ids: "___fmp__.RepeatedString" = aristaproto.message_field(8)
"""list of child assignments"""
created_at: datetime = aristaproto.message_field(9)
"""created_at is the time when the ConfigletAssignment was created."""
created_by: Optional[str] = aristaproto.message_field(
10, wraps=aristaproto.TYPE_STRING
)
"""created_by is the user who created the ConfigletAssignment."""
last_modified_at: datetime = aristaproto.message_field(11)
"""
last_modified_at is the time when the ConfigletAssignment
was last modified.
"""
last_modified_by: Optional[str] = aristaproto.message_field(
12, wraps=aristaproto.TYPE_STRING
)
"""
last_modified_by is the user who last modified the ConfigletAssignment.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletRequest(aristaproto.Message):
""" """
key: "ConfigletKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a Configlet 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 ConfigletResponse(aristaproto.Message):
""" """
value: "Configlet" = 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
Configlet instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletKey"] = 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 ConfigletSomeResponse(aristaproto.Message):
""" """
value: "Configlet" = 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)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Configlet"] = 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.
"""
filter: "Filter" = aristaproto.message_field(2)
"""
For each Configlet in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given Configlet is streamed to the user.
"""
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 Configlet at end.
* Each Configlet response is fully-specified (all fields set).
* start: Returns the state of each Configlet at start, followed by updates until now.
* Each Configlet response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Configlet at start, followed by updates
until end.
* Each Configlet response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletStreamResponse(aristaproto.Message):
""" """
value: "Configlet" = 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 Configlet's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the Configlet 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 ConfigletBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["Configlet"] = 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.
"""
filter: "Filter" = aristaproto.message_field(2)
"""
For each Configlet in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given Configlet is streamed to the user.
"""
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 Configlet at end.
* Each Configlet response is fully-specified (all fields set).
* start: Returns the state of each Configlet at start, followed by updates until now.
* Each Configlet response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Configlet at start, followed by updates
until end.
* Each Configlet response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
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 ConfigletBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigletStreamResponse"] = 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 ConfigletAssignmentRequest(aristaproto.Message):
""" """
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a ConfigletAssignment 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 ConfigletAssignmentResponse(aristaproto.Message):
""" """
value: "ConfigletAssignment" = 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
ConfigletAssignment instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletAssignmentKey"] = 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 ConfigletAssignmentSomeResponse(aristaproto.Message):
""" """
value: "ConfigletAssignment" = 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)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletAssignment"] = 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 ConfigletAssignment at end.
* Each ConfigletAssignment response is fully-specified (all fields set).
* start: Returns the state of each ConfigletAssignment at start, followed by updates until now.
* Each ConfigletAssignment response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletAssignment at start, followed by updates
until end.
* Each ConfigletAssignment response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentStreamResponse(aristaproto.Message):
""" """
value: "ConfigletAssignment" = 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 ConfigletAssignment's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the ConfigletAssignment 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 ConfigletAssignmentBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletAssignment"] = 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 ConfigletAssignment at end.
* Each ConfigletAssignment response is fully-specified (all fields set).
* start: Returns the state of each ConfigletAssignment at start, followed by updates until now.
* Each ConfigletAssignment response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletAssignment at start, followed by updates
until end.
* Each ConfigletAssignment response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
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 ConfigletAssignmentBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigletAssignmentStreamResponse"] = 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 ConfigletAssignmentConfigRequest(aristaproto.Message):
""" """
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a ConfigletAssignmentConfig 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 ConfigletAssignmentConfigResponse(aristaproto.Message):
""" """
value: "ConfigletAssignmentConfig" = 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
ConfigletAssignmentConfig instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletAssignmentKey"] = 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 ConfigletAssignmentConfigSomeResponse(aristaproto.Message):
""" """
value: "ConfigletAssignmentConfig" = 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)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletAssignmentConfig"] = 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 ConfigletAssignmentConfig at end.
* Each ConfigletAssignmentConfig response is fully-specified (all fields set).
* start: Returns the state of each ConfigletAssignmentConfig at start, followed by updates until now.
* Each ConfigletAssignmentConfig response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletAssignmentConfig at start, followed by updates
until end.
* Each ConfigletAssignmentConfig response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigStreamResponse(aristaproto.Message):
""" """
value: "ConfigletAssignmentConfig" = 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 ConfigletAssignmentConfig's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the ConfigletAssignmentConfig 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 ConfigletAssignmentConfigBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletAssignmentConfig"] = 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 ConfigletAssignmentConfig at end.
* Each ConfigletAssignmentConfig response is fully-specified (all fields set).
* start: Returns the state of each ConfigletAssignmentConfig at start, followed by updates until now.
* Each ConfigletAssignmentConfig response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletAssignmentConfig at start, followed by updates
until end.
* Each ConfigletAssignmentConfig response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
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 ConfigletAssignmentConfigBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigletAssignmentConfigStreamResponse"] = (
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 ConfigletAssignmentConfigSetRequest(aristaproto.Message):
""" """
value: "ConfigletAssignmentConfig" = aristaproto.message_field(1)
"""
ConfigletAssignmentConfig carries the value to set into the datastore.
See the documentation on the ConfigletAssignmentConfig struct for which fields are required.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigSetResponse(aristaproto.Message):
""" """
value: "ConfigletAssignmentConfig" = aristaproto.message_field(1)
"""
Value carries all the values given in the ConfigletAssignmentConfigSetRequest as well
as any server-generated values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the (UTC) timestamp at which the system recognizes the
creation. The only guarantees made about this timestamp are:
- it is after the time the request was received
- a time-ranged query with StartTime==CreatedAt will include this instance.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigSetSomeRequest(aristaproto.Message):
""" """
values: List["ConfigletAssignmentConfig"] = aristaproto.message_field(1)
"""
value contains a list of ConfigletAssignmentConfig values to write.
It is possible to provide more values than can fit within either:
- the maxiumum send size of the client
- the maximum receive size of the server
If this error occurs you must reduce the number of values sent.
See gRPC \"maximum message size\" documentation for more information.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigSetSomeResponse(aristaproto.Message):
""" """
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
"""
error: str = aristaproto.string_field(2)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteRequest(aristaproto.Message):
""" """
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
Key indicates which ConfigletAssignmentConfig instance to remove.
This field must always be set.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteResponse(aristaproto.Message):
""" """
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
Key echoes back the key of the deleted ConfigletAssignmentConfig instance.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the (UTC) timestamp at which the system recognizes the
deletion. The only guarantees made about this timestamp are:
- it is after the time the request was received
- a time-ranged query with StartTime==DeletedAt will not include this instance.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletAssignmentKey"] = aristaproto.message_field(1)
"""key contains a list of ConfigletAssignmentConfig keys to delete"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteSomeResponse(aristaproto.Message):
"""
ConfigletAssignmentConfigDeleteSomeResponse is only sent when there is an error.
"""
key: "ConfigletAssignmentKey" = aristaproto.message_field(1)
"""
"""
error: str = aristaproto.string_field(2)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteAllRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletAssignmentConfig"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a DeleteAll.
This requires all provided fields to be equal to the response.
A filtered DeleteAll will use GetAll with filter to find things to delete.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletAssignmentConfigDeleteAllResponse(aristaproto.Message):
""" """
type: "___fmp__.DeleteError" = aristaproto.enum_field(1)
"""
This describes the class of delete error.
A DeleteAllResponse is only sent when there is an error.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""This indicates the error message from the delete failure."""
key: "ConfigletAssignmentKey" = aristaproto.message_field(3)
"""
This is the key of the ConfigletAssignmentConfig instance that failed to be deleted.
"""
time: datetime = aristaproto.message_field(4)
"""Time indicates the (UTC) timestamp when the key was being deleted."""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigRequest(aristaproto.Message):
""" """
key: "ConfigletKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a ConfigletConfig 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 ConfigletConfigResponse(aristaproto.Message):
""" """
value: "ConfigletConfig" = 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
ConfigletConfig instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletKey"] = 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 ConfigletConfigSomeResponse(aristaproto.Message):
""" """
value: "ConfigletConfig" = 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)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletConfig"] = 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.
"""
filter: "Filter" = aristaproto.message_field(2)
"""
For each ConfigletConfig in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given ConfigletConfig is streamed to the user.
"""
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 ConfigletConfig at end.
* Each ConfigletConfig response is fully-specified (all fields set).
* start: Returns the state of each ConfigletConfig at start, followed by updates until now.
* Each ConfigletConfig response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletConfig at start, followed by updates
until end.
* Each ConfigletConfig response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigStreamResponse(aristaproto.Message):
""" """
value: "ConfigletConfig" = 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 ConfigletConfig's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the ConfigletConfig 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 ConfigletConfigBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletConfig"] = 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.
"""
filter: "Filter" = aristaproto.message_field(2)
"""
For each ConfigletConfig in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given ConfigletConfig is streamed to the user.
"""
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 ConfigletConfig at end.
* Each ConfigletConfig response is fully-specified (all fields set).
* start: Returns the state of each ConfigletConfig at start, followed by updates until now.
* Each ConfigletConfig response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each ConfigletConfig at start, followed by updates
until end.
* Each ConfigletConfig response at start is fully-specified, but updates until end may
be partial.
This field is not allowed in the Subscribe RPC.
"""
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 ConfigletConfigBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["ConfigletConfigStreamResponse"] = 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 ConfigletConfigSetRequest(aristaproto.Message):
""" """
value: "ConfigletConfig" = aristaproto.message_field(1)
"""
ConfigletConfig carries the value to set into the datastore.
See the documentation on the ConfigletConfig struct for which fields are required.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigSetResponse(aristaproto.Message):
""" """
value: "ConfigletConfig" = aristaproto.message_field(1)
"""
Value carries all the values given in the ConfigletConfigSetRequest as well
as any server-generated values.
"""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the (UTC) timestamp at which the system recognizes the
creation. The only guarantees made about this timestamp are:
- it is after the time the request was received
- a time-ranged query with StartTime==CreatedAt will include this instance.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigSetSomeRequest(aristaproto.Message):
""" """
values: List["ConfigletConfig"] = aristaproto.message_field(1)
"""
value contains a list of ConfigletConfig values to write.
It is possible to provide more values than can fit within either:
- the maxiumum send size of the client
- the maximum receive size of the server
If this error occurs you must reduce the number of values sent.
See gRPC \"maximum message size\" documentation for more information.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigSetSomeResponse(aristaproto.Message):
""" """
key: "ConfigletKey" = aristaproto.message_field(1)
"""
"""
error: str = aristaproto.string_field(2)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteRequest(aristaproto.Message):
""" """
key: "ConfigletKey" = aristaproto.message_field(1)
"""
Key indicates which ConfigletConfig instance to remove.
This field must always be set.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteResponse(aristaproto.Message):
""" """
key: "ConfigletKey" = aristaproto.message_field(1)
"""Key echoes back the key of the deleted ConfigletConfig instance."""
time: datetime = aristaproto.message_field(2)
"""
Time indicates the (UTC) timestamp at which the system recognizes the
deletion. The only guarantees made about this timestamp are:
- it is after the time the request was received
- a time-ranged query with StartTime==DeletedAt will not include this instance.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteSomeRequest(aristaproto.Message):
""" """
keys: List["ConfigletKey"] = aristaproto.message_field(1)
"""key contains a list of ConfigletConfig keys to delete"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteSomeResponse(aristaproto.Message):
"""
ConfigletConfigDeleteSomeResponse is only sent when there is an error.
"""
key: "ConfigletKey" = aristaproto.message_field(1)
"""
"""
error: str = aristaproto.string_field(2)
"""
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteAllRequest(aristaproto.Message):
""" """
partial_eq_filter: List["ConfigletConfig"] = aristaproto.message_field(1)
"""
PartialEqFilter provides a way to server-side filter a DeleteAll.
This requires all provided fields to be equal to the response.
A filtered DeleteAll will use GetAll with filter to find things to delete.
"""
filter: "Filter" = aristaproto.message_field(2)
"""
For each ConfigletConfig in the list, all populated fields are considered ANDed together
as a filtering operation. Similarly, the list itself is ORed such that any individual
filter that matches a given ConfigletConfig will be deleted.
"""
[docs]
@dataclass(eq=False, repr=False)
class ConfigletConfigDeleteAllResponse(aristaproto.Message):
""" """
type: "___fmp__.DeleteError" = aristaproto.enum_field(1)
"""
This describes the class of delete error.
A DeleteAllResponse is only sent when there is an error.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""This indicates the error message from the delete failure."""
key: "ConfigletKey" = aristaproto.message_field(3)
"""
This is the key of the ConfigletConfig instance that failed to be deleted.
"""
time: datetime = aristaproto.message_field(4)
"""Time indicates the (UTC) timestamp when the key was being deleted."""
[docs]
class ConfigletServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
configlet_request: "ConfigletRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletService/GetOne",
configlet_request,
ConfigletResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
configlet_some_request: "ConfigletSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletService/GetSome",
configlet_some_request,
ConfigletSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
configlet_stream_request: "ConfigletStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletService/GetAll",
configlet_stream_request,
ConfigletStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
configlet_stream_request: "ConfigletStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletService/Subscribe",
configlet_stream_request,
ConfigletStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
configlet_batched_stream_request: "ConfigletBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletService/GetAllBatched",
configlet_batched_stream_request,
ConfigletBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
configlet_batched_stream_request: "ConfigletBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletService/SubscribeBatched",
configlet_batched_stream_request,
ConfigletBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class ConfigletAssignmentServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
configlet_assignment_request: "ConfigletAssignmentRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletAssignmentResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletAssignmentService/GetOne",
configlet_assignment_request,
ConfigletAssignmentResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
configlet_assignment_some_request: "ConfigletAssignmentSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentService/GetSome",
configlet_assignment_some_request,
ConfigletAssignmentSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
configlet_assignment_stream_request: "ConfigletAssignmentStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentService/GetAll",
configlet_assignment_stream_request,
ConfigletAssignmentStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
configlet_assignment_stream_request: "ConfigletAssignmentStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentService/Subscribe",
configlet_assignment_stream_request,
ConfigletAssignmentStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
configlet_assignment_batched_stream_request: "ConfigletAssignmentBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentService/GetAllBatched",
configlet_assignment_batched_stream_request,
ConfigletAssignmentBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
configlet_assignment_batched_stream_request: "ConfigletAssignmentBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentService/SubscribeBatched",
configlet_assignment_batched_stream_request,
ConfigletAssignmentBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class ConfigletAssignmentConfigServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
configlet_assignment_config_request: "ConfigletAssignmentConfigRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletAssignmentConfigResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetOne",
configlet_assignment_config_request,
ConfigletAssignmentConfigResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
configlet_assignment_config_some_request: "ConfigletAssignmentConfigSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetSome",
configlet_assignment_config_some_request,
ConfigletAssignmentConfigSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
configlet_assignment_config_stream_request: "ConfigletAssignmentConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetAll",
configlet_assignment_config_stream_request,
ConfigletAssignmentConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
configlet_assignment_config_stream_request: "ConfigletAssignmentConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/Subscribe",
configlet_assignment_config_stream_request,
ConfigletAssignmentConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def set(
self,
configlet_assignment_config_set_request: "ConfigletAssignmentConfigSetRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletAssignmentConfigSetResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletAssignmentConfigService/Set",
configlet_assignment_config_set_request,
ConfigletAssignmentConfigSetResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def set_some(
self,
configlet_assignment_config_set_some_request: "ConfigletAssignmentConfigSetSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigSetSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/SetSome",
configlet_assignment_config_set_some_request,
ConfigletAssignmentConfigSetSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete(
self,
configlet_assignment_config_delete_request: "ConfigletAssignmentConfigDeleteRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletAssignmentConfigDeleteResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletAssignmentConfigService/Delete",
configlet_assignment_config_delete_request,
ConfigletAssignmentConfigDeleteResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def delete_some(
self,
configlet_assignment_config_delete_some_request: "ConfigletAssignmentConfigDeleteSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigDeleteSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/DeleteSome",
configlet_assignment_config_delete_some_request,
ConfigletAssignmentConfigDeleteSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete_all(
self,
configlet_assignment_config_delete_all_request: "ConfigletAssignmentConfigDeleteAllRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigDeleteAllResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/DeleteAll",
configlet_assignment_config_delete_all_request,
ConfigletAssignmentConfigDeleteAllResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
configlet_assignment_config_batched_stream_request: "ConfigletAssignmentConfigBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetAllBatched",
configlet_assignment_config_batched_stream_request,
ConfigletAssignmentConfigBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
configlet_assignment_config_batched_stream_request: "ConfigletAssignmentConfigBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletAssignmentConfigBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletAssignmentConfigService/SubscribeBatched",
configlet_assignment_config_batched_stream_request,
ConfigletAssignmentConfigBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class ConfigletConfigServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
configlet_config_request: "ConfigletConfigRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletConfigResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletConfigService/GetOne",
configlet_config_request,
ConfigletConfigResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
configlet_config_some_request: "ConfigletConfigSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/GetSome",
configlet_config_some_request,
ConfigletConfigSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
configlet_config_stream_request: "ConfigletConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/GetAll",
configlet_config_stream_request,
ConfigletConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
configlet_config_stream_request: "ConfigletConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/Subscribe",
configlet_config_stream_request,
ConfigletConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def set(
self,
configlet_config_set_request: "ConfigletConfigSetRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletConfigSetResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletConfigService/Set",
configlet_config_set_request,
ConfigletConfigSetResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def set_some(
self,
configlet_config_set_some_request: "ConfigletConfigSetSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigSetSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/SetSome",
configlet_config_set_some_request,
ConfigletConfigSetSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete(
self,
configlet_config_delete_request: "ConfigletConfigDeleteRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "ConfigletConfigDeleteResponse":
""" """
return await self._unary_unary(
"/arista.configlet.v1.ConfigletConfigService/Delete",
configlet_config_delete_request,
ConfigletConfigDeleteResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def delete_some(
self,
configlet_config_delete_some_request: "ConfigletConfigDeleteSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigDeleteSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/DeleteSome",
configlet_config_delete_some_request,
ConfigletConfigDeleteSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete_all(
self,
configlet_config_delete_all_request: "ConfigletConfigDeleteAllRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigDeleteAllResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/DeleteAll",
configlet_config_delete_all_request,
ConfigletConfigDeleteAllResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
configlet_config_batched_stream_request: "ConfigletConfigBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/GetAllBatched",
configlet_config_batched_stream_request,
ConfigletConfigBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
configlet_config_batched_stream_request: "ConfigletConfigBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[ConfigletConfigBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.configlet.v1.ConfigletConfigService/SubscribeBatched",
configlet_config_batched_stream_request,
ConfigletConfigBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
from .... import fmp as ___fmp__
from ... import subscriptions as __subscriptions__
from ... import time as __time__
[docs]
class ConfigletServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, configlet_request: "ConfigletRequest"
) -> "ConfigletResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, configlet_some_request: "ConfigletSomeRequest"
) -> AsyncIterator[ConfigletSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, configlet_stream_request: "ConfigletStreamRequest"
) -> AsyncIterator[ConfigletStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, configlet_stream_request: "ConfigletStreamRequest"
) -> AsyncIterator[ConfigletStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self, configlet_batched_stream_request: "ConfigletBatchedStreamRequest"
) -> AsyncIterator[ConfigletBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self, configlet_batched_stream_request: "ConfigletBatchedStreamRequest"
) -> AsyncIterator[ConfigletBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[ConfigletRequest, ConfigletResponse]"
) -> 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[ConfigletSomeRequest, ConfigletSomeResponse]",
) -> 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[ConfigletStreamRequest, ConfigletStreamResponse]",
) -> 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[ConfigletStreamRequest, ConfigletStreamResponse]",
) -> 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[ConfigletStreamRequest, 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[ConfigletStreamRequest, 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[ConfigletBatchedStreamRequest, ConfigletBatchedStreamResponse]",
) -> 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[ConfigletBatchedStreamRequest, ConfigletBatchedStreamResponse]",
) -> 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.configlet.v1.ConfigletService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletRequest,
ConfigletResponse,
),
"/arista.configlet.v1.ConfigletService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletSomeRequest,
ConfigletSomeResponse,
),
"/arista.configlet.v1.ConfigletService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletStreamRequest,
ConfigletStreamResponse,
),
"/arista.configlet.v1.ConfigletService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletStreamRequest,
ConfigletStreamResponse,
),
"/arista.configlet.v1.ConfigletService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletBatchedStreamRequest,
ConfigletBatchedStreamResponse,
),
"/arista.configlet.v1.ConfigletService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletBatchedStreamRequest,
ConfigletBatchedStreamResponse,
),
}
[docs]
class ConfigletAssignmentServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, configlet_assignment_request: "ConfigletAssignmentRequest"
) -> "ConfigletAssignmentResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, configlet_assignment_some_request: "ConfigletAssignmentSomeRequest"
) -> AsyncIterator[ConfigletAssignmentSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, configlet_assignment_stream_request: "ConfigletAssignmentStreamRequest"
) -> AsyncIterator[ConfigletAssignmentStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, configlet_assignment_stream_request: "ConfigletAssignmentStreamRequest"
) -> AsyncIterator[ConfigletAssignmentStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
configlet_assignment_batched_stream_request: "ConfigletAssignmentBatchedStreamRequest",
) -> AsyncIterator[ConfigletAssignmentBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
configlet_assignment_batched_stream_request: "ConfigletAssignmentBatchedStreamRequest",
) -> AsyncIterator[ConfigletAssignmentBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentRequest, ConfigletAssignmentResponse]",
) -> 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[ConfigletAssignmentSomeRequest, ConfigletAssignmentSomeResponse]",
) -> 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[ConfigletAssignmentStreamRequest, ConfigletAssignmentStreamResponse]",
) -> 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[ConfigletAssignmentStreamRequest, ConfigletAssignmentStreamResponse]",
) -> 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[ConfigletAssignmentStreamRequest, 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[ConfigletAssignmentStreamRequest, 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[ConfigletAssignmentBatchedStreamRequest, ConfigletAssignmentBatchedStreamResponse]",
) -> 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[ConfigletAssignmentBatchedStreamRequest, ConfigletAssignmentBatchedStreamResponse]",
) -> 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.configlet.v1.ConfigletAssignmentService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentRequest,
ConfigletAssignmentResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentSomeRequest,
ConfigletAssignmentSomeResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentStreamRequest,
ConfigletAssignmentStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentStreamRequest,
ConfigletAssignmentStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentBatchedStreamRequest,
ConfigletAssignmentBatchedStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentBatchedStreamRequest,
ConfigletAssignmentBatchedStreamResponse,
),
}
[docs]
class ConfigletAssignmentConfigServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, configlet_assignment_config_request: "ConfigletAssignmentConfigRequest"
) -> "ConfigletAssignmentConfigResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self,
configlet_assignment_config_some_request: "ConfigletAssignmentConfigSomeRequest",
) -> AsyncIterator[ConfigletAssignmentConfigSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self,
configlet_assignment_config_stream_request: "ConfigletAssignmentConfigStreamRequest",
) -> AsyncIterator[ConfigletAssignmentConfigStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self,
configlet_assignment_config_stream_request: "ConfigletAssignmentConfigStreamRequest",
) -> AsyncIterator[ConfigletAssignmentConfigStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set(
self,
configlet_assignment_config_set_request: "ConfigletAssignmentConfigSetRequest",
) -> "ConfigletAssignmentConfigSetResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set_some(
self,
configlet_assignment_config_set_some_request: "ConfigletAssignmentConfigSetSomeRequest",
) -> AsyncIterator[ConfigletAssignmentConfigSetSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete(
self,
configlet_assignment_config_delete_request: "ConfigletAssignmentConfigDeleteRequest",
) -> "ConfigletAssignmentConfigDeleteResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_some(
self,
configlet_assignment_config_delete_some_request: "ConfigletAssignmentConfigDeleteSomeRequest",
) -> AsyncIterator[ConfigletAssignmentConfigDeleteSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_all(
self,
configlet_assignment_config_delete_all_request: "ConfigletAssignmentConfigDeleteAllRequest",
) -> AsyncIterator[ConfigletAssignmentConfigDeleteAllResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
configlet_assignment_config_batched_stream_request: "ConfigletAssignmentConfigBatchedStreamRequest",
) -> AsyncIterator[ConfigletAssignmentConfigBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
configlet_assignment_config_batched_stream_request: "ConfigletAssignmentConfigBatchedStreamRequest",
) -> AsyncIterator[ConfigletAssignmentConfigBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigRequest, ConfigletAssignmentConfigResponse]",
) -> 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[ConfigletAssignmentConfigSomeRequest, ConfigletAssignmentConfigSomeResponse]",
) -> 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[ConfigletAssignmentConfigStreamRequest, ConfigletAssignmentConfigStreamResponse]",
) -> 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[ConfigletAssignmentConfigStreamRequest, ConfigletAssignmentConfigStreamResponse]",
) -> 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[ConfigletAssignmentConfigStreamRequest, 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[ConfigletAssignmentConfigStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_set(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigSetRequest, ConfigletAssignmentConfigSetResponse]",
) -> None:
request = await stream.recv_message()
response = await self.set(request)
await stream.send_message(response)
async def __rpc_set_some(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigSetSomeRequest, ConfigletAssignmentConfigSetSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.set_some,
stream,
request,
)
async def __rpc_delete(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigDeleteRequest, ConfigletAssignmentConfigDeleteResponse]",
) -> None:
request = await stream.recv_message()
response = await self.delete(request)
await stream.send_message(response)
async def __rpc_delete_some(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigDeleteSomeRequest, ConfigletAssignmentConfigDeleteSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.delete_some,
stream,
request,
)
async def __rpc_delete_all(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigDeleteAllRequest, ConfigletAssignmentConfigDeleteAllResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.delete_all,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[ConfigletAssignmentConfigBatchedStreamRequest, ConfigletAssignmentConfigBatchedStreamResponse]",
) -> 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[ConfigletAssignmentConfigBatchedStreamRequest, ConfigletAssignmentConfigBatchedStreamResponse]",
) -> 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.configlet.v1.ConfigletAssignmentConfigService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentConfigRequest,
ConfigletAssignmentConfigResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigSomeRequest,
ConfigletAssignmentConfigSomeResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigStreamRequest,
ConfigletAssignmentConfigStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigStreamRequest,
ConfigletAssignmentConfigStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentConfigStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/Set": grpclib.const.Handler(
self.__rpc_set,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentConfigSetRequest,
ConfigletAssignmentConfigSetResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/SetSome": grpclib.const.Handler(
self.__rpc_set_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigSetSomeRequest,
ConfigletAssignmentConfigSetSomeResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/Delete": grpclib.const.Handler(
self.__rpc_delete,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletAssignmentConfigDeleteRequest,
ConfigletAssignmentConfigDeleteResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/DeleteSome": grpclib.const.Handler(
self.__rpc_delete_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigDeleteSomeRequest,
ConfigletAssignmentConfigDeleteSomeResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/DeleteAll": grpclib.const.Handler(
self.__rpc_delete_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigDeleteAllRequest,
ConfigletAssignmentConfigDeleteAllResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigBatchedStreamRequest,
ConfigletAssignmentConfigBatchedStreamResponse,
),
"/arista.configlet.v1.ConfigletAssignmentConfigService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletAssignmentConfigBatchedStreamRequest,
ConfigletAssignmentConfigBatchedStreamResponse,
),
}
[docs]
class ConfigletConfigServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, configlet_config_request: "ConfigletConfigRequest"
) -> "ConfigletConfigResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, configlet_config_some_request: "ConfigletConfigSomeRequest"
) -> AsyncIterator[ConfigletConfigSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, configlet_config_stream_request: "ConfigletConfigStreamRequest"
) -> AsyncIterator[ConfigletConfigStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, configlet_config_stream_request: "ConfigletConfigStreamRequest"
) -> AsyncIterator[ConfigletConfigStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set(
self, configlet_config_set_request: "ConfigletConfigSetRequest"
) -> "ConfigletConfigSetResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set_some(
self, configlet_config_set_some_request: "ConfigletConfigSetSomeRequest"
) -> AsyncIterator[ConfigletConfigSetSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete(
self, configlet_config_delete_request: "ConfigletConfigDeleteRequest"
) -> "ConfigletConfigDeleteResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_some(
self, configlet_config_delete_some_request: "ConfigletConfigDeleteSomeRequest"
) -> AsyncIterator[ConfigletConfigDeleteSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_all(
self, configlet_config_delete_all_request: "ConfigletConfigDeleteAllRequest"
) -> AsyncIterator[ConfigletConfigDeleteAllResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
configlet_config_batched_stream_request: "ConfigletConfigBatchedStreamRequest",
) -> AsyncIterator[ConfigletConfigBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
configlet_config_batched_stream_request: "ConfigletConfigBatchedStreamRequest",
) -> AsyncIterator[ConfigletConfigBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[ConfigletConfigRequest, ConfigletConfigResponse]",
) -> 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[ConfigletConfigSomeRequest, ConfigletConfigSomeResponse]",
) -> 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[ConfigletConfigStreamRequest, ConfigletConfigStreamResponse]",
) -> 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[ConfigletConfigStreamRequest, ConfigletConfigStreamResponse]",
) -> 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[ConfigletConfigStreamRequest, 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[ConfigletConfigStreamRequest, MetaResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
async def __rpc_set(
self,
stream: "grpclib.server.Stream[ConfigletConfigSetRequest, ConfigletConfigSetResponse]",
) -> None:
request = await stream.recv_message()
response = await self.set(request)
await stream.send_message(response)
async def __rpc_set_some(
self,
stream: "grpclib.server.Stream[ConfigletConfigSetSomeRequest, ConfigletConfigSetSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.set_some,
stream,
request,
)
async def __rpc_delete(
self,
stream: "grpclib.server.Stream[ConfigletConfigDeleteRequest, ConfigletConfigDeleteResponse]",
) -> None:
request = await stream.recv_message()
response = await self.delete(request)
await stream.send_message(response)
async def __rpc_delete_some(
self,
stream: "grpclib.server.Stream[ConfigletConfigDeleteSomeRequest, ConfigletConfigDeleteSomeResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.delete_some,
stream,
request,
)
async def __rpc_delete_all(
self,
stream: "grpclib.server.Stream[ConfigletConfigDeleteAllRequest, ConfigletConfigDeleteAllResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.delete_all,
stream,
request,
)
async def __rpc_get_all_batched(
self,
stream: "grpclib.server.Stream[ConfigletConfigBatchedStreamRequest, ConfigletConfigBatchedStreamResponse]",
) -> 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[ConfigletConfigBatchedStreamRequest, ConfigletConfigBatchedStreamResponse]",
) -> 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.configlet.v1.ConfigletConfigService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletConfigRequest,
ConfigletConfigResponse,
),
"/arista.configlet.v1.ConfigletConfigService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigSomeRequest,
ConfigletConfigSomeResponse,
),
"/arista.configlet.v1.ConfigletConfigService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigStreamRequest,
ConfigletConfigStreamResponse,
),
"/arista.configlet.v1.ConfigletConfigService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigStreamRequest,
ConfigletConfigStreamResponse,
),
"/arista.configlet.v1.ConfigletConfigService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletConfigStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletConfigService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigStreamRequest,
MetaResponse,
),
"/arista.configlet.v1.ConfigletConfigService/Set": grpclib.const.Handler(
self.__rpc_set,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletConfigSetRequest,
ConfigletConfigSetResponse,
),
"/arista.configlet.v1.ConfigletConfigService/SetSome": grpclib.const.Handler(
self.__rpc_set_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigSetSomeRequest,
ConfigletConfigSetSomeResponse,
),
"/arista.configlet.v1.ConfigletConfigService/Delete": grpclib.const.Handler(
self.__rpc_delete,
grpclib.const.Cardinality.UNARY_UNARY,
ConfigletConfigDeleteRequest,
ConfigletConfigDeleteResponse,
),
"/arista.configlet.v1.ConfigletConfigService/DeleteSome": grpclib.const.Handler(
self.__rpc_delete_some,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigDeleteSomeRequest,
ConfigletConfigDeleteSomeResponse,
),
"/arista.configlet.v1.ConfigletConfigService/DeleteAll": grpclib.const.Handler(
self.__rpc_delete_all,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigDeleteAllRequest,
ConfigletConfigDeleteAllResponse,
),
"/arista.configlet.v1.ConfigletConfigService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigBatchedStreamRequest,
ConfigletConfigBatchedStreamResponse,
),
"/arista.configlet.v1.ConfigletConfigService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
ConfigletConfigBatchedStreamRequest,
ConfigletConfigBatchedStreamResponse,
),
}