hermes-chatto-plugin/vendor/chattolib/_pb/chatto/admin/v1/diagnostics_connect.py

# -*- coding: utf-8 -*-
# Generated by https://github.com/connectrpc/connect-python.  DO NOT EDIT!
# source: chatto/admin/v1/diagnostics.proto

from collections.abc import AsyncGenerator, AsyncIterator, Iterable, Iterator, Mapping
from typing import Protocol

from connectrpc.client import ConnectClient, ConnectClientSync
from connectrpc.code import Code
from connectrpc.compression import Compression
from connectrpc.errors import ConnectError
from connectrpc.interceptor import Interceptor, InterceptorSync
from connectrpc.method import IdempotencyLevel, MethodInfo
from connectrpc.request import Headers, RequestContext
from connectrpc.server import ConnectASGIApplication, ConnectWSGIApplication, Endpoint, EndpointSync
import chatto.admin.v1.diagnostics_pb2 as chatto_dot_admin_dot_v1_dot_diagnostics__pb2


class AdminDiagnosticsService(Protocol):
    async def get_system_info(self, request: chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class AdminDiagnosticsServiceASGIApplication(ConnectASGIApplication[AdminDiagnosticsService]):
    def __init__(self, service: AdminDiagnosticsService | AsyncGenerator[AdminDiagnosticsService], *, interceptors: Iterable[Interceptor]=(), read_max_bytes: int | None = None, compressions: Iterable[Compression] | None = None) -> None:
        super().__init__(
            service=service,
            endpoints=lambda svc: {
                "/chatto.admin.v1.AdminDiagnosticsService/GetSystemInfo": Endpoint.unary(
                    method=MethodInfo(
                        name="GetSystemInfo",
                        service_name="chatto.admin.v1.AdminDiagnosticsService",
                        input=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
                        output=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.get_system_info,
                ),
            },
            interceptors=interceptors,
            read_max_bytes=read_max_bytes,
            compressions=compressions,
        )

    @property
    def path(self) -> str:
        """Returns the URL path to mount the application to when serving multiple applications."""
        return "/chatto.admin.v1.AdminDiagnosticsService"


class AdminDiagnosticsServiceClient(ConnectClient):
    async def get_system_info(
        self,
        request: chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetSystemInfo",
                service_name="chatto.admin.v1.AdminDiagnosticsService",
                input=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
                output=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )


class AdminDiagnosticsServiceSync(Protocol):
    def get_system_info(self, request: chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class AdminDiagnosticsServiceWSGIApplication(ConnectWSGIApplication):
    def __init__(self, service: AdminDiagnosticsServiceSync, interceptors: Iterable[InterceptorSync]=(), read_max_bytes: int | None = None, compressions: Iterable[Compression] | None = None) -> None:
        super().__init__(
            endpoints={
                "/chatto.admin.v1.AdminDiagnosticsService/GetSystemInfo": EndpointSync.unary(
                    method=MethodInfo(
                        name="GetSystemInfo",
                        service_name="chatto.admin.v1.AdminDiagnosticsService",
                        input=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
                        output=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.get_system_info,
                ),
            },
            interceptors=interceptors,
            read_max_bytes=read_max_bytes,
            compressions=compressions,
        )

    @property
    def path(self) -> str:
        """Returns the URL path to mount the application to when serving multiple applications."""
        return "/chatto.admin.v1.AdminDiagnosticsService"


class AdminDiagnosticsServiceClientSync(ConnectClientSync):
    def get_system_info(
        self,
        request: chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetSystemInfo",
                service_name="chatto.admin.v1.AdminDiagnosticsService",
                input=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoRequest,
                output=chatto_dot_admin_dot_v1_dot_diagnostics__pb2.GetSystemInfoResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )