# Copyright (c) 2026 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/logging.v1/logging.proto, arista/logging.v1/services.gen.proto
# plugin: python-aristaproto
# This file has been @generated
__all__ = (
"LoggingCategory",
"LoggingLevel",
"LoggingKey",
"LoggingConfig",
"LoggingAudit",
"Logging",
"MetaResponse",
"LoggingRequest",
"LoggingResponse",
"LoggingSomeRequest",
"LoggingSomeResponse",
"LoggingStreamRequest",
"LoggingStreamResponse",
"LoggingConfigRequest",
"LoggingConfigResponse",
"LoggingConfigSomeRequest",
"LoggingConfigSomeResponse",
"LoggingConfigStreamRequest",
"LoggingConfigStreamResponse",
"LoggingConfigSetRequest",
"LoggingConfigSetResponse",
"LoggingConfigSetSomeRequest",
"LoggingConfigSetSomeResponse",
"LoggingConfigDeleteRequest",
"LoggingConfigDeleteResponse",
"LoggingConfigDeleteSomeRequest",
"LoggingConfigDeleteSomeResponse",
"LoggingConfigDeleteAllRequest",
"LoggingConfigDeleteAllResponse",
"LoggingServiceStub",
"LoggingServiceBase",
"LoggingConfigServiceStub",
"LoggingConfigServiceBase",
)
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 LoggingCategory(aristaproto.Enum):
"""LoggingCategory describes the possible categories of logging."""
UNSPECIFIED = 0
"""
LOGGING_CATEGORY_UNSPECIFIED is an unspecified category. Disallowed for writes.
"""
CHANGECONTROL = 1
"""
LOGGING_CATEGORY_CHANGECONTROL is used for actions executed as part of a change control.
"""
ACTION_EXEC = 2
"""
LOGGING_CATEGORY_ACTION_EXEC is used for actions executed via autofill actions, build hooks,
or the standalone execution feature.
"""
STUDIO_TEMPLATE = 3
"""
LOGGING_CATEGORY_STUDIO_TEMPLATE is used for studio template rendering executions.
"""
[docs]
class LoggingLevel(aristaproto.Enum):
"""LoggingLevel describes the different verbosity levels of logging."""
UNSPECIFIED = 0
"""
LOGGING_LEVEL_UNSPECIFIED is an unspecified level. Disallowed for writes,
but setting this in a GetAll filter returns all logging levels.
"""
TRACE = 1
"""LOGGING_LEVEL_TRACE is equivalent to a trace verbosity log call."""
DEBUG = 2
"""LOGGING_LEVEL_DEBUG is equivalent to a debug verbosity log call."""
INFO = 3
"""LOGGING_LEVEL_INFO is equivalent to an info verbosity log call."""
WARNING = 4
"""LOGGING_LEVEL_WARNING is equivalent to a warning verbosity log call."""
ERROR = 5
"""LOGGING_LEVEL_ERROR is equivalent to an error verbosity log call."""
CRITICAL = 6
"""
LOGGING_LEVEL_CRITICAL is equivalent to a critical verbosity log call.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingKey(aristaproto.Message):
"""
LoggingKey uniquely identifies a log via its id, its category and level.
All log lines for executions of change controls, studios, etc. are written under the same key.
"""
id: Optional[str] = aristaproto.message_field(1, wraps=aristaproto.TYPE_STRING)
"""id is a unique identifier provided by the client."""
category: "LoggingCategory" = aristaproto.enum_field(2)
"""category is the category of the log."""
level: "LoggingLevel" = aristaproto.enum_field(3)
"""level is the verbosity level of the log."""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfig(aristaproto.Message):
"""
LoggingConfig is used to write non-audit related logs produced by such services as
change controls, studio builds, python autofill actions, etc. by providing a key
related to the current execution, the log, and optional metadata such as
file name, line number and tags.
"""
key: "LoggingKey" = aristaproto.message_field(1)
"""
key uniquely identifies a standalone log or a series of logs for a single execution.
"""
log: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""log contains the actual log contents."""
file: Optional[str] = aristaproto.message_field(3, wraps=aristaproto.TYPE_STRING)
"""
file contains the file name of the script being executed, if relevant.
"""
line: Optional[int] = aristaproto.message_field(4, wraps=aristaproto.TYPE_INT64)
"""
line contains the line number of the log in the script being executed, if relevant.
"""
tags: "___fmp__.MapStringString" = aristaproto.message_field(5)
"""
tags is a map of tag values that can be set on write and can be used in queries
to filter the logs. Some executions will automatically populate some tags. E.g.
change control logs will populate a \"changeControlId\" and \"stageId\" tag.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingAudit(aristaproto.Message):
"""
LoggingAudit contains audit information about the given log. It's added by the logging rAPI
and only accessible in the state model.
"""
added_by: Optional[str] = aristaproto.message_field(1, wraps=aristaproto.TYPE_STRING)
"""added_by contains the name of the user who added the log."""
[docs]
@dataclass(eq=False, repr=False)
class Logging(aristaproto.Message):
"""
Logging is used to retrieve non-audit related logs produced by such services as
change controls, studio builds, python autofill actions, etc.
"""
key: "LoggingKey" = aristaproto.message_field(1)
"""
key uniquely identifies a standalone log or a series of logs for a single execution.
"""
log: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""log contains the actual log contents."""
file: Optional[str] = aristaproto.message_field(3, wraps=aristaproto.TYPE_STRING)
"""
file contains the file name of the script being executed, if relevant.
"""
line: Optional[int] = aristaproto.message_field(4, wraps=aristaproto.TYPE_INT64)
"""
line contains the line number of the log in the script being executed, if relevant.
"""
audit: "LoggingAudit" = aristaproto.message_field(5)
"""
audit contains additional audit information added by the service when the log was
written.
"""
tags: "___fmp__.MapStringString" = aristaproto.message_field(6)
"""
tags is a map of tag values that was set on write and can be used in queries
to filter the logs. Some executions will automatically populate some tags. E.g.
change control logs will populate a \"changeControlId\" and \"stageId\" tag.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingRequest(aristaproto.Message):
key: "LoggingKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a Logging 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 LoggingResponse(aristaproto.Message):
value: "Logging" = 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
Logging instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingSomeRequest(aristaproto.Message):
keys: List["LoggingKey"] = 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 LoggingSomeResponse(aristaproto.Message):
value: "Logging" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
Logging instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingStreamRequest(aristaproto.Message):
partial_eq_filter: List["Logging"] = 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 Logging at end.
* Each Logging response is fully-specified (all fields set).
* start: Returns the state of each Logging at start, followed by updates until now.
* Each Logging response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each Logging at start, followed by updates
until end.
* Each Logging response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingStreamResponse(aristaproto.Message):
value: "Logging" = 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 Logging's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the Logging 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 LoggingConfigRequest(aristaproto.Message):
key: "LoggingKey" = aristaproto.message_field(1)
"""
Key uniquely identifies a LoggingConfig 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 LoggingConfigResponse(aristaproto.Message):
value: "LoggingConfig" = 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
LoggingConfig instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigSomeRequest(aristaproto.Message):
keys: List["LoggingKey"] = 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 LoggingConfigSomeResponse(aristaproto.Message):
value: "LoggingConfig" = aristaproto.message_field(1)
"""
Value is the value requested.
This structure will be fully-populated as it exists in the datastore. If
optional fields were not given at creation, these fields will be empty or
set to default values.
"""
error: Optional[str] = aristaproto.message_field(2, wraps=aristaproto.TYPE_STRING)
"""
Error is an optional field.
It should be filled when there is an error in the GetSome process.
"""
time: datetime = aristaproto.message_field(3)
"""
Time carries the (UTC) timestamp of the last-modification of the
LoggingConfig instance in this response.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigStreamRequest(aristaproto.Message):
partial_eq_filter: List["LoggingConfig"] = 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 LoggingConfig at end.
* Each LoggingConfig response is fully-specified (all fields set).
* start: Returns the state of each LoggingConfig at start, followed by updates until now.
* Each LoggingConfig response at start is fully-specified, but updates may be partial.
* start and end: Returns the state of each LoggingConfig at start, followed by updates
until end.
* Each LoggingConfig response at start is fully-specified, but updates until end may
be partial.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigStreamResponse(aristaproto.Message):
value: "LoggingConfig" = 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 LoggingConfig's last modification."""
type: "__subscriptions__.Operation" = aristaproto.enum_field(3)
"""
Operation indicates how the LoggingConfig 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 LoggingConfigSetRequest(aristaproto.Message):
value: "LoggingConfig" = aristaproto.message_field(1)
"""
LoggingConfig carries the value to set into the datastore.
See the documentation on the LoggingConfig struct for which fields are required.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigSetResponse(aristaproto.Message):
value: "LoggingConfig" = aristaproto.message_field(1)
"""
Value carries all the values given in the LoggingConfigSetRequest 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 LoggingConfigSetSomeRequest(aristaproto.Message):
values: List["LoggingConfig"] = aristaproto.message_field(1)
"""
value contains a list of LoggingConfig 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 LoggingConfigSetSomeResponse(aristaproto.Message):
key: "LoggingKey" = aristaproto.message_field(1)
error: str = aristaproto.string_field(2)
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigDeleteRequest(aristaproto.Message):
key: "LoggingKey" = aristaproto.message_field(1)
"""
Key indicates which LoggingConfig instance to remove.
This field must always be set.
"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigDeleteResponse(aristaproto.Message):
key: "LoggingKey" = aristaproto.message_field(1)
"""Key echoes back the key of the deleted LoggingConfig 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 LoggingConfigDeleteSomeRequest(aristaproto.Message):
keys: List["LoggingKey"] = aristaproto.message_field(1)
"""key contains a list of LoggingConfig keys to delete"""
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigDeleteSomeResponse(aristaproto.Message):
"""LoggingConfigDeleteSomeResponse is only sent when there is an error."""
key: "LoggingKey" = aristaproto.message_field(1)
error: str = aristaproto.string_field(2)
[docs]
@dataclass(eq=False, repr=False)
class LoggingConfigDeleteAllRequest(aristaproto.Message):
partial_eq_filter: List["LoggingConfig"] = 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 LoggingConfigDeleteAllResponse(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: "LoggingKey" = aristaproto.message_field(3)
"""
This is the key of the LoggingConfig 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 LoggingServiceStub(aristaproto.ServiceStub):
[docs]
async def get_one(
self,
logging_request: "LoggingRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "LoggingResponse":
return await self._unary_unary(
"/arista.logging.v1.LoggingService/GetOne",
logging_request,
LoggingResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
logging_some_request: "LoggingSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingSomeResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingService/GetSome",
logging_some_request,
LoggingSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
logging_stream_request: "LoggingStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingStreamResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingService/GetAll",
logging_stream_request,
LoggingStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
logging_stream_request: "LoggingStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingStreamResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingService/Subscribe",
logging_stream_request,
LoggingStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
class LoggingConfigServiceStub(aristaproto.ServiceStub):
[docs]
async def get_one(
self,
logging_config_request: "LoggingConfigRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "LoggingConfigResponse":
return await self._unary_unary(
"/arista.logging.v1.LoggingConfigService/GetOne",
logging_config_request,
LoggingConfigResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def get_some(
self,
logging_config_some_request: "LoggingConfigSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigSomeResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/GetSome",
logging_config_some_request,
LoggingConfigSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def get_all(
self,
logging_config_stream_request: "LoggingConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigStreamResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/GetAll",
logging_config_stream_request,
LoggingConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def subscribe(
self,
logging_config_stream_request: "LoggingConfigStreamRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigStreamResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/Subscribe",
logging_config_stream_request,
LoggingConfigStreamResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def set(
self,
logging_config_set_request: "LoggingConfigSetRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "LoggingConfigSetResponse":
return await self._unary_unary(
"/arista.logging.v1.LoggingConfigService/Set",
logging_config_set_request,
LoggingConfigSetResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def set_some(
self,
logging_config_set_some_request: "LoggingConfigSetSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigSetSomeResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/SetSome",
logging_config_set_some_request,
LoggingConfigSetSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete(
self,
logging_config_delete_request: "LoggingConfigDeleteRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "LoggingConfigDeleteResponse":
return await self._unary_unary(
"/arista.logging.v1.LoggingConfigService/Delete",
logging_config_delete_request,
LoggingConfigDeleteResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def delete_some(
self,
logging_config_delete_some_request: "LoggingConfigDeleteSomeRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigDeleteSomeResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/DeleteSome",
logging_config_delete_some_request,
LoggingConfigDeleteSomeResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
):
yield response
[docs]
async def delete_all(
self,
logging_config_delete_all_request: "LoggingConfigDeleteAllRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "AsyncIterator[LoggingConfigDeleteAllResponse]":
async for response in self._unary_stream(
"/arista.logging.v1.LoggingConfigService/DeleteAll",
logging_config_delete_all_request,
LoggingConfigDeleteAllResponse,
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 LoggingServiceBase(ServiceBase):
[docs]
async def get_one(self, logging_request: "LoggingRequest") -> "LoggingResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, logging_some_request: "LoggingSomeRequest"
) -> AsyncIterator[LoggingSomeResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, logging_stream_request: "LoggingStreamRequest"
) -> AsyncIterator[LoggingStreamResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, logging_stream_request: "LoggingStreamRequest"
) -> AsyncIterator[LoggingStreamResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[LoggingRequest, LoggingResponse]"
) -> 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[LoggingSomeRequest, LoggingSomeResponse]"
) -> 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[LoggingStreamRequest, LoggingStreamResponse]"
) -> 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[LoggingStreamRequest, LoggingStreamResponse]"
) -> 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[LoggingStreamRequest, 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[LoggingStreamRequest, MetaResponse]"
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.subscribe_meta,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.logging.v1.LoggingService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingRequest,
LoggingResponse,
),
"/arista.logging.v1.LoggingService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingSomeRequest,
LoggingSomeResponse,
),
"/arista.logging.v1.LoggingService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingStreamRequest,
LoggingStreamResponse,
),
"/arista.logging.v1.LoggingService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingStreamRequest,
LoggingStreamResponse,
),
"/arista.logging.v1.LoggingService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingStreamRequest,
MetaResponse,
),
"/arista.logging.v1.LoggingService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingStreamRequest,
MetaResponse,
),
}
[docs]
class LoggingConfigServiceBase(ServiceBase):
[docs]
async def get_one(
self, logging_config_request: "LoggingConfigRequest"
) -> "LoggingConfigResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_some(
self, logging_config_some_request: "LoggingConfigSomeRequest"
) -> AsyncIterator[LoggingConfigSomeResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def get_all(
self, logging_config_stream_request: "LoggingConfigStreamRequest"
) -> AsyncIterator[LoggingConfigStreamResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def subscribe(
self, logging_config_stream_request: "LoggingConfigStreamRequest"
) -> AsyncIterator[LoggingConfigStreamResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set(
self, logging_config_set_request: "LoggingConfigSetRequest"
) -> "LoggingConfigSetResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def set_some(
self, logging_config_set_some_request: "LoggingConfigSetSomeRequest"
) -> AsyncIterator[LoggingConfigSetSomeResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete(
self, logging_config_delete_request: "LoggingConfigDeleteRequest"
) -> "LoggingConfigDeleteResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_some(
self, logging_config_delete_some_request: "LoggingConfigDeleteSomeRequest"
) -> AsyncIterator[LoggingConfigDeleteSomeResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def delete_all(
self, logging_config_delete_all_request: "LoggingConfigDeleteAllRequest"
) -> AsyncIterator[LoggingConfigDeleteAllResponse]:
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_get_one(
self, stream: "grpclib.server.Stream[LoggingConfigRequest, LoggingConfigResponse]"
) -> 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[LoggingConfigSomeRequest, LoggingConfigSomeResponse]"
) -> 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[LoggingConfigStreamRequest, LoggingConfigStreamResponse]",
) -> 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[LoggingConfigStreamRequest, LoggingConfigStreamResponse]",
) -> 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[LoggingConfigStreamRequest, 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[LoggingConfigStreamRequest, 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[LoggingConfigSetRequest, LoggingConfigSetResponse]"
) -> 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[LoggingConfigSetSomeRequest, LoggingConfigSetSomeResponse]",
) -> 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[LoggingConfigDeleteRequest, LoggingConfigDeleteResponse]",
) -> 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[LoggingConfigDeleteSomeRequest, LoggingConfigDeleteSomeResponse]",
) -> 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[LoggingConfigDeleteAllRequest, LoggingConfigDeleteAllResponse]",
) -> None:
request = await stream.recv_message()
await self._call_rpc_handler_server_stream(
self.delete_all,
stream,
request,
)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/arista.logging.v1.LoggingConfigService/GetOne": grpclib.const.Handler(
self.__rpc_get_one,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingConfigRequest,
LoggingConfigResponse,
),
"/arista.logging.v1.LoggingConfigService/GetSome": grpclib.const.Handler(
self.__rpc_get_some,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigSomeRequest,
LoggingConfigSomeResponse,
),
"/arista.logging.v1.LoggingConfigService/GetAll": grpclib.const.Handler(
self.__rpc_get_all,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigStreamRequest,
LoggingConfigStreamResponse,
),
"/arista.logging.v1.LoggingConfigService/Subscribe": grpclib.const.Handler(
self.__rpc_subscribe,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigStreamRequest,
LoggingConfigStreamResponse,
),
"/arista.logging.v1.LoggingConfigService/GetMeta": grpclib.const.Handler(
self.__rpc_get_meta,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingConfigStreamRequest,
MetaResponse,
),
"/arista.logging.v1.LoggingConfigService/SubscribeMeta": grpclib.const.Handler(
self.__rpc_subscribe_meta,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigStreamRequest,
MetaResponse,
),
"/arista.logging.v1.LoggingConfigService/Set": grpclib.const.Handler(
self.__rpc_set,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingConfigSetRequest,
LoggingConfigSetResponse,
),
"/arista.logging.v1.LoggingConfigService/SetSome": grpclib.const.Handler(
self.__rpc_set_some,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigSetSomeRequest,
LoggingConfigSetSomeResponse,
),
"/arista.logging.v1.LoggingConfigService/Delete": grpclib.const.Handler(
self.__rpc_delete,
grpclib.const.Cardinality.UNARY_UNARY,
LoggingConfigDeleteRequest,
LoggingConfigDeleteResponse,
),
"/arista.logging.v1.LoggingConfigService/DeleteSome": grpclib.const.Handler(
self.__rpc_delete_some,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigDeleteSomeRequest,
LoggingConfigDeleteSomeResponse,
),
"/arista.logging.v1.LoggingConfigService/DeleteAll": grpclib.const.Handler(
self.__rpc_delete_all,
grpclib.const.Cardinality.UNARY_STREAM,
LoggingConfigDeleteAllRequest,
LoggingConfigDeleteAllResponse,
),
}