# 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/endpointlocation.v1/endpointlocation.proto, arista/endpointlocation.v1/services.gen.proto
# plugin: python-aristaproto
# This file has been @generated
__all__ = (
"MacType",
"Likelihood",
"IdentifierType",
"IdentifierSource",
"Explanation",
"DeviceType",
"DeviceStatus",
"IdentifierSourceList",
"Identifier",
"IdentifierList",
"ExplanationList",
"Location",
"LocationList",
"DeviceInfo",
"Device",
"DeviceMap",
"EndpointLocationKey",
"EndpointLocation",
"MetaResponse",
"EndpointLocationRequest",
"EndpointLocationResponse",
"EndpointLocationSomeRequest",
"EndpointLocationSomeResponse",
"EndpointLocationStreamRequest",
"EndpointLocationStreamResponse",
"EndpointLocationBatchedStreamRequest",
"EndpointLocationBatchedStreamResponse",
"EndpointLocationServiceStub",
"EndpointLocationServiceBase",
)
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 MacType(aristaproto.Enum):
"""
MacType describes how the MAC address was learned on the port location.
"""
UNSPECIFIED = 0
"""MAC_TYPE_UNSPECIFIED is the default unspecified MAC type."""
LEARNED_DYNAMIC = 1
"""
MAC_TYPE_LEARNED_DYNAMIC indicates a MAC that is dynamically learned in hardware.
"""
LEARNED_SECURE = 2
"""
MAC_TYPE_LEARNED_SECURE indicates a MAC that is learned in hardware on a port-security protect interface.
"""
CONFIGURED_DYNAMIC = 3
"""MAC_TYPE_CONFIGURED_DYNAMIC indicates a configured dynamic MAC."""
CONFIGURED_SECURE = 4
"""
MAC_TYPE_CONFIGURED_SECURE indicates a MAC configured on an interface with Port Security: Protect mode enabled
"""
CONFIGURED_STATIC = 5
"""MAC_TYPE_CONFIGURED_STATIC indicates a statically configured MAC."""
PEER_DYNAMIC = 6
"""
MAC_TYPE_PEER_DYNAMIC indicates a dynamically learned MAC discovered via an MLAG peer.
"""
PEER_STATIC = 7
"""
MAC_TYPE_PEER_STATIC indicates a statically configured MAC discovered via an MLAG peer.
"""
PEER_SECURE = 8
"""
MAC_TYPE_PEER_SECURE indicates a secure MAC learned from an MLAG peer.
"""
LEARNED_REMOTE = 9
"""
MAC_TYPE_LEARNED_REMOTE indicates a MAC learned remotely behind a VxLAN VTEP.
"""
CONFIGURED_REMOTE = 10
"""
MAC_TYPE_CONFIGURED_REMOTE indicates a MAC configured behind a VxLAN VTEP.
"""
RECEIVED_REMOTE = 11
"""
MAC_TYPE_RECEIVED_REMOTE indicates a MAC received from a VxLAN controller.
"""
PEER_LEARNED_REMOTE = 12
"""
MAC_TYPE_PEER_LEARNED_REMOTE indicates a remote MAC learned on a peer VTEP.
"""
PEER_CONFIGURED_REMOTE = 13
"""
MAC_TYPE_PEER_CONFIGURED_REMOTE indicates an MLAG peer configured remote MAC.
"""
PEER_RECEIVED_REMOTE = 14
"""
MAC_TYPE_PEER_RECEIVED_REMOTE indicates a remote MAC received from a VxLAN controller by the MLAG peer.
"""
EVPN_DYNAMIC_REMOTE = 15
"""MAC_TYPE_EVPN_DYNAMIC_REMOTE indicates an EVPN dynamic remote MAC."""
EVPN_CONFIGURED_REMOTE = 16
"""
MAC_TYPE_EVPN_CONFIGURED_REMOTE indicates an EVPN configured remote MAC.
"""
PEER_EVPN_REMOTE = 17
"""MAC_TYPE_PEER_EVPN_REMOTE indicates an MLAG peer EVPN remote MAC."""
CONFIGURED_ROUTER = 18
"""
MAC_TYPE_CONFIGURED_ROUTER indicates a configured MAC used in routing.
"""
PEER_ROUTER = 19
"""MAC_TYPE_PEER_ROUTER indicates an MLAG peer specific router MAC."""
EVPN_INTF_DYNAMIC = 20
"""
MAC_TYPE_EVPN_INTF_DYNAMIC indicates a MAC advertised by EVPN when a dynamic MAC is learned on ESI (Ethernet Segment Identifier).
"""
EVPN_INTF_STATIC = 21
"""MAC_TYPE_EVPN_INTF_STATIC indicates a MAC configured on EVPN ESI."""
AUTHENTICATED = 22
"""MAC_TYPE_AUTHENTICATED indicates a MAC authenticated via 802.1X."""
PEER_AUTHENTICATED = 23
"""
MAC_TYPE_PEER_AUTHENTICATED indicates a MAC authenticated by 802.1X and learned on an MLAG peer.
"""
PENDING_SECURE = 24
"""MAC_TYPE_PENDING_SECURE indicates a secure MAC in a pending state."""
SOFTWARE_LEARNED_DYNAMIC = 25
"""
MAC_TYPE_SOFTWARE_LEARNED_DYNAMIC indicates a MAC learned behind VTEP in software in the case of a VxLAN remote MAC.
"""
OTHER = 99999
"""MAC_TYPE_OTHER is used for capturing future MAC types."""
[docs]
class Likelihood(aristaproto.Enum):
"""Likelihood indicates a level of confidence."""
UNSPECIFIED = 0
"""LIKELIHOOD_UNSPECIFIED is the default unspecified likelihood."""
VERY_LIKELY = 1
"""LIKELIHOOD_VERY_LIKELY indicates very high confidence."""
LIKELY = 2
"""LIKELIHOOD_LIKELY indicates high confidence."""
SOMEWHAT_LIKELY = 3
"""LIKELIHOOD_SOMEWHAT_LIKELY indicates medium confidence."""
LESS_LIKELY = 4
"""LIKELIHOOD_LESS_LIKELY indicates low confidence."""
[docs]
class IdentifierType(aristaproto.Enum):
"""IdentifierType defines the set of ways for identifying endpoints."""
UNSPECIFIED = 0
"""IDENTIFIER_TYPE_UNSPECIFIED is the default unspecified identifier."""
MAC_ADDR = 1
"""IDENTIFIER_TYPE_MAC_ADDR indicates a MAC address identifier."""
IPV4_ADDR = 2
"""IDENTIFIER_TYPE_IPV4_ADDR indicates an IPv4 address identifier."""
IPV6_ADDR = 3
"""IDENTIFIER_TYPE_IPV6_ADDR indicates an IPv6 address identifier."""
INVENTORY_DEVICE_ID = 4
"""
IDENTIFIER_TYPE_INVENTORY_DEVICE_ID indicates an inventory device identifier.
"""
PRIMARY_MANAGEMENT_IP = 5
"""
IDENTIFIER_TYPE_PRIMARY_MANAGEMENT_IP indicates a primary management IP identifier.
"""
HOSTNAME = 6
"""IDENTIFIER_TYPE_HOSTNAME indicates a hostname identifier."""
USERNAME = 7
"""
IDENTIFIER_TYPE_USERNAME indicates a username identifier.
Deprecated - do not return usernames of wifi endpoints anymore
"""
OTHER = 99999
"""IDENTIFIER_TYPE_OTHER is used for an unknown identifier."""
[docs]
class IdentifierSource(aristaproto.Enum):
"""
IdentifierSource defines the set of network protocols and other
information sources where an identifier was found.
"""
UNSPECIFIED = 0
"""
IDENTIFIER_SOURCE_UNSPECIFIED is the default unspecified identifier source.
"""
FDB = 1
"""IDENTIFIER_SOURCE_FDB indicates a forwarding table."""
ARP = 2
"""IDENTIFIER_SOURCE_ARP indicates ARP (IPv4)."""
NEIGHBOR = 3
"""IDENTIFIER_SOURCE_NEIGHBOR indicates NDP (IPv6)."""
DEVICE_INVENTORY = 4
"""
IDENTIFIER_SOURCE_DEVICE_INVENTORY indicates that an endpoint is in the
CloudVision inventory.
"""
LLDP = 5
"""IDENTIFIER_SOURCE_LLDP indicates LLDP."""
DHCP = 6
"""IDENTIFIER_SOURCE_DHCP indicates DHCP."""
WIFI = 7
"""
IDENTIFIER_SOURCE_WIFI indicates a WiFi endpoint that was learned through
a wireless manager.
Deprecated - do not support searches for wifi endpoints anymore
"""
[docs]
class Explanation(aristaproto.Enum):
"""
Explanation defines the set of reasons for why a location has a certain likelihood.
Some explanations apply to queried endpoints while others apply to queried devices
in the CloudVision device inventory.
"""
UNSPECIFIED = 0
"""EXPLANATION_UNSPECIFIED is the default unspecified explanation."""
DIRECT_CONNECTION = 1
"""
EXPLANATION_DIRECT_CONNECTION indicates a direct connection to the queried
endpoint device discovered via LLDP.
"""
NON_INVENTORY_CONNECTION = 2
"""
EXPLANATION_NON_INVENTORY_CONNECTION indicates a direct connection to at least
one non-inventory device (likely discovered via LLDP).
"""
NO_CONNECTION = 3
"""
EXPLANATION_NO_CONNECTION indicates that a location has no direct connection to
any known device.
"""
INVENTORY_CONNECTION = 4
"""
EXPLANATION_INVENTORY_CONNECTION indicates that all of a location's known connections
are to devices in the inventory.
"""
OWN_PORT_INVENTORY_DEVICE = 5
"""
EXPLANATION_OWN_PORT_INVENTORY_DEVICE indicates a location on the queried inventory
device itself.
"""
DIRECT_CONNECTION_INVENTORY_DEVICE = 6
"""
EXPLANATION_DIRECT_CONNECTION_INVENTORY_DEVICE indicates a direct connection to the
queried inventory device discovered via LLDP.
"""
NO_CONNECTION_INVENTORY_DEVICE = 7
"""
EXPLANATION_NO_CONNECTION_INVENTORY_DEVICE indicates that a location has no direct
connection to any known device (for queried inventory devices).
"""
OTHER_CONNECTION_INVENTORY_DEVICE = 8
"""
EXPLANATION_OTHER_CONNECTION_INVENTORY_DEVICE indicates that a location only has
connections to other devices besides the queried inventory device.
"""
VIRTUAL = 9
"""
EXPLANATION_VIRTUAL indicates that a location contains a virtual interface.
"""
WIRELESS_CONNECTION = 10
"""
EXPLANATION_WIRELESS_CONNECTION indicates a direct connection from a wireless access
point in the device inventory to a wireless client.
"""
ACCESS_PORT = 11
"""EXPLANATION_ACCESS_PORT indicates that a location is an access port."""
[docs]
class DeviceType(aristaproto.Enum):
"""
DeviceType defines a broad set of categories for
all queried devices.
"""
UNSPECIFIED = 0
"""DEVICE_TYPE_UNSPECIFIED is the default unspecified device type."""
INVENTORY = 1
"""
DEVICE_TYPE_INVENTORY indicates a device in the CloudVision inventory.
"""
ENDPOINT = 2
"""
DEVICE_TYPE_ENDPOINT indicates an endpoint that does not exist
in the CloudVision inventory.
"""
WIFI_ENDPOINT = 3
"""
DEVICE_TYPE_WIFI_ENDPOINT indicates a WiFi client/endpoint that does not
exist in the CloudVision inventory.
Deprecated - do not support searches for wifi endpoints anymore
"""
[docs]
class DeviceStatus(aristaproto.Enum):
"""DeviceStatus is the network status of a device."""
UNSPECIFIED = 0
"""DEVICE_STATUS_UNSPECIFIED is the default unspecified device status."""
ACTIVE = 1
"""
DEVICE_STATUS_ACTIVE indicates a device is streaming its telemetry data
to CloudVision.
"""
INACTIVE = 2
"""
DEVICE_STATUS_INACTIVE indicates a device is either not streaming its
telemetry data to CloudVision or has been decommissioned from CloudVision.
"""
[docs]
@dataclass(eq=False, repr=False)
class IdentifierSourceList(aristaproto.Message):
"""IdentifierSourceList is a list of IdentifierSource."""
values: List["IdentifierSource"] = aristaproto.enum_field(1)
"""values is an unordered list of unique IdentifierSource."""
[docs]
@dataclass(eq=False, repr=False)
class Identifier(aristaproto.Message):
"""Identifier holds device identification information."""
type: "IdentifierType" = aristaproto.enum_field(1)
"""type is the identifier type corresponding to value."""
value: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
value is the string representation of the identifier. Its interpretation
depends on type.
"""
source_list: "IdentifierSourceList" = aristaproto.message_field(3)
"""
source_list is the set of sources where this identifier was discovered.
"""
[docs]
@dataclass(eq=False, repr=False)
class IdentifierList(aristaproto.Message):
"""IdentifierList is a list of Identifier."""
values: List["Identifier"] = aristaproto.message_field(1)
"""
values is an unordered list of Identifier where each Identifier has a
unique type and value combination.
"""
[docs]
@dataclass(eq=False, repr=False)
class ExplanationList(aristaproto.Message):
"""ExplanationList is a list of Explanation."""
values: List["Explanation"] = aristaproto.enum_field(1)
"""
values is a list of unique Explanation. Currently, this will
always contain one value, but in the future it may contain
more.
"""
[docs]
@dataclass(eq=False, repr=False)
class Location(aristaproto.Message):
"""
Location is a port (device_id, interface, vlan_id) on which
at least one identifier has been discovered.
"""
device_id: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""device_id identifies the device of the port."""
device_status: "DeviceStatus" = aristaproto.enum_field(2)
"""device_status is the status of the device identified by device_id."""
interface: Optional[str] = aristaproto.message_field(
3, wraps=aristaproto.TYPE_STRING
)
"""interface is the interface of the port."""
vlan_id: Optional[int] = aristaproto.message_field(4, wraps=aristaproto.TYPE_UINT32)
"""vlan_id identifies the VLAN of the port."""
learned_time: datetime = aristaproto.message_field(5)
"""learned_time is when the port learned its identifiers."""
mac_type: "MacType" = aristaproto.enum_field(6)
"""mac_type is how the port learned its MAC address identifier."""
likelihood: "Likelihood" = aristaproto.enum_field(7)
"""
likelihood is the probability level that the port is directly connected
to the queried endpoint.
"""
explanation_list: "ExplanationList" = aristaproto.message_field(8)
"""
explanation_list holds the reasons that the port was assigned likelihood.
"""
identifier_list: "IdentifierList" = aristaproto.message_field(9)
"""identifier_list holds the discovered identifiers of the port."""
[docs]
@dataclass(eq=False, repr=False)
class LocationList(aristaproto.Message):
"""LocationList is a list of Location."""
values: List["Location"] = aristaproto.message_field(1)
"""values is a list of unique Location."""
[docs]
@dataclass(eq=False, repr=False)
class DeviceInfo(aristaproto.Message):
"""
DeviceInfo holds various attributes of a device (typically an endpoint) from Fingerbank.
"""
device_name: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""device_name is the name of the device."""
mobile: Optional[bool] = aristaproto.message_field(2, wraps=aristaproto.TYPE_BOOL)
"""mobile indicates whether the device is a mobile."""
tablet: Optional[bool] = aristaproto.message_field(3, wraps=aristaproto.TYPE_BOOL)
"""tablet indicates whether the device is a tablet."""
score: Optional[int] = aristaproto.message_field(4, wraps=aristaproto.TYPE_UINT32)
"""
score is a value from 0 to 100 that indicates how confident we are
that the device has device_name.
Fingerbank API documentation of score: https://api.fingerbank.org/api_doc/2/combinations.html
"""
version: Optional[str] = aristaproto.message_field(5, wraps=aristaproto.TYPE_STRING)
"""version is the version of device_name."""
mac_vendor: Optional[str] = aristaproto.message_field(
6, wraps=aristaproto.TYPE_STRING
)
"""
mac_vendor is the enterprise that assigns the MAC address of the device.
"""
classification: Optional[str] = aristaproto.message_field(
7, wraps=aristaproto.TYPE_STRING
)
"""
classification is the broadest category to which device_name belongs.
This is the highest level in hierarchy.
"""
hierarchy: "___fmp__.RepeatedString" = aristaproto.message_field(8)
"""
hierarchy is a list of categorizations of the device from most broad
to most specific. The first element is always classification and the
last element is always device_name.
For example,
[\"VoIP Device\", \"FooInc VoIP\", \"FooInc PhoneSet IP\", \"FooInc PhoneSet IP Model123\"]
In this hierarchy, \"VoIP Device\" is classification and \"FooInc PhoneSet IP Model123\"
is device_name.
"""
[docs]
@dataclass(eq=False, repr=False)
class Device(aristaproto.Message):
"""Device holds information for a device matching a search term."""
identifier_list: "IdentifierList" = aristaproto.message_field(1)
"""identifier_list holds the unique identifiers for the device."""
device_type: "DeviceType" = aristaproto.enum_field(2)
"""device_type is the broad category of the device."""
location_list: "LocationList" = aristaproto.message_field(3)
"""
location_list is the list of possible locations of the device.
It is ordered from most likely to least likely.
"""
device_status: "DeviceStatus" = aristaproto.enum_field(4)
"""device_status is the network status of the device."""
device_info: "DeviceInfo" = aristaproto.message_field(5)
"""device_info holds various attributes of the device."""
[docs]
@dataclass(eq=False, repr=False)
class DeviceMap(aristaproto.Message):
"""DeviceMap is a collection of Device."""
values: Dict[str, "Device"] = aristaproto.map_field(
1, aristaproto.TYPE_STRING, aristaproto.TYPE_MESSAGE
)
"""
values is a map from most specific identifier to Device.
The key could be device serial number or MAC address.
"""
[docs]
@dataclass(eq=False, repr=False)
class EndpointLocationKey(aristaproto.Message):
"""EndpointLocationKey holds a search term used to locate an endpoint."""
search_term: Optional[str] = aristaproto.message_field(
1, wraps=aristaproto.TYPE_STRING
)
"""
search_term is used to match against identifiers on devices.
This could be a MAC/IP address, hostname, etc.
"""
[docs]
@dataclass(eq=False, repr=False)
class EndpointLocation(aristaproto.Message):
"""
EndpointLocation is the entrypoint to searching for
endpoints.
"""
key: "EndpointLocationKey" = aristaproto.message_field(1)
"""key holds a search term used to locate an endpoint."""
device_map: "DeviceMap" = aristaproto.message_field(2)
"""
device_map holds the devices (and their potential locations)
that match the search term.
"""
[docs]
@dataclass(eq=False, repr=False)
class EndpointLocationRequest(aristaproto.Message):
""" """
key: "EndpointLocationKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a EndpointLocation 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 EndpointLocationResponse(aristaproto.Message):
""" """
value: "EndpointLocation" = 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
EndpointLocation instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class EndpointLocationSomeRequest(aristaproto.Message):
""" """
keys: List["EndpointLocationKey"] = 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 EndpointLocationSomeResponse(aristaproto.Message):
""" """
value: "EndpointLocation" = 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 EndpointLocationStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["EndpointLocation"] = 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 EndpointLocation at end.
* Each EndpointLocation response is fully-specified (all fields set).
* start: Returns the state of each EndpointLocation at start, followed by updates until now.
* Each EndpointLocation response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each EndpointLocation at start, followed by updates
until end.
* Each EndpointLocation 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 EndpointLocationStreamResponse(aristaproto.Message):
""" """
value: "EndpointLocation" = 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 EndpointLocation's last modification.
"""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the EndpointLocation 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 EndpointLocationBatchedStreamRequest(aristaproto.Message):
""" """
partial_eq_filter: List["EndpointLocation"] = 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 EndpointLocation at end.
* Each EndpointLocation response is fully-specified (all fields set).
* start: Returns the state of each EndpointLocation at start, followed by updates until now.
* Each EndpointLocation response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each EndpointLocation at start, followed by updates
until end.
* Each EndpointLocation 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 EndpointLocationBatchedStreamResponse(aristaproto.Message):
""" """
responses: List["EndpointLocationStreamResponse"] = aristaproto.message_field(1)
"""
Values are the values deemed relevant to the initiating request.
The length of this structure is guaranteed to be between (inclusive) 1 and
min(req.max_messages, INTERNAL_BATCH_LIMIT).
"""
[docs]
class EndpointLocationServiceStub(aristaproto.ServiceStub):
""" """
[docs]
async def get_one(
self,
endpoint_location_request: "EndpointLocationRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "EndpointLocationResponse":
""" """
return await self._unary_unary(
"/arista.endpointlocation.v1.EndpointLocationService/GetOne",
endpoint_location_request,
EndpointLocationResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
endpoint_location_some_request: "EndpointLocationSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[EndpointLocationSomeResponse]":
""" """
async for response in self._unary_stream(
"/arista.endpointlocation.v1.EndpointLocationService/GetSome",
endpoint_location_some_request,
EndpointLocationSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
endpoint_location_stream_request: "EndpointLocationStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[EndpointLocationStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.endpointlocation.v1.EndpointLocationService/GetAll",
endpoint_location_stream_request,
EndpointLocationStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
endpoint_location_stream_request: "EndpointLocationStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[EndpointLocationStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.endpointlocation.v1.EndpointLocationService/Subscribe",
endpoint_location_stream_request,
EndpointLocationStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all_batched(
self,
endpoint_location_batched_stream_request: "EndpointLocationBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[EndpointLocationBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.endpointlocation.v1.EndpointLocationService/GetAllBatched",
endpoint_location_batched_stream_request,
EndpointLocationBatchedStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe_batched(
self,
endpoint_location_batched_stream_request: "EndpointLocationBatchedStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[EndpointLocationBatchedStreamResponse]":
""" """
async for response in self._unary_stream(
"/arista.endpointlocation.v1.EndpointLocationService/SubscribeBatched",
endpoint_location_batched_stream_request,
EndpointLocationBatchedStreamResponse,
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 EndpointLocationServiceBase(ServiceBase):
""" """
[docs]
async def get_one(
self, endpoint_location_request: "EndpointLocationRequest"
) -> "EndpointLocationResponse":
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, endpoint_location_some_request: "EndpointLocationSomeRequest"
) -> AsyncIterator[EndpointLocationSomeResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, endpoint_location_stream_request: "EndpointLocationStreamRequest"
) -> AsyncIterator[EndpointLocationStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, endpoint_location_stream_request: "EndpointLocationStreamRequest"
) -> AsyncIterator[EndpointLocationStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all_batched(
self,
endpoint_location_batched_stream_request: "EndpointLocationBatchedStreamRequest",
) -> AsyncIterator[EndpointLocationBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe_batched(
self,
endpoint_location_batched_stream_request: "EndpointLocationBatchedStreamRequest",
) -> AsyncIterator[EndpointLocationBatchedStreamResponse]:
""" """
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self,
stream: "grpclib.server.Stream[EndpointLocationRequest, EndpointLocationResponse]",
) -> 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[EndpointLocationSomeRequest, EndpointLocationSomeResponse]",
) -> 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[EndpointLocationStreamRequest, EndpointLocationStreamResponse]",
) -> 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[EndpointLocationStreamRequest, EndpointLocationStreamResponse]",
) -> 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[EndpointLocationStreamRequest, 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[EndpointLocationStreamRequest, 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[EndpointLocationBatchedStreamRequest, EndpointLocationBatchedStreamResponse]",
) -> 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[EndpointLocationBatchedStreamRequest, EndpointLocationBatchedStreamResponse]",
) -> 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.endpointlocation.v1.EndpointLocationService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
EndpointLocationRequest,
EndpointLocationResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationSomeRequest,
EndpointLocationSomeResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationStreamRequest,
EndpointLocationStreamResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationStreamRequest,
EndpointLocationStreamResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
EndpointLocationStreamRequest,
MetaResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationStreamRequest,
MetaResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/GetAllBatched": grpclib.const.Handler(
self.__rpc_get_all_batched,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationBatchedStreamRequest,
EndpointLocationBatchedStreamResponse,
),
"/arista.endpointlocation.v1.EndpointLocationService/SubscribeBatched": grpclib.const.Handler(
self.__rpc_subscribe_batched,
grpclib.const.Cardinality.UNARY_STREAM,
EndpointLocationBatchedStreamRequest,
EndpointLocationBatchedStreamResponse,
),
}