# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: keyapis/iot_connect/v1/keyapis_iot_connect_manage_v1.proto, keyapis/iot_connect/v1/keyapis_iot_connect_system_v1.proto
# plugin: python-betterproto
# This file has been @generated

from typing import TYPE_CHECKING

import betterproto
import grpclib
from betterproto.grpc.grpclib_server import ServiceBase
from pydantic import model_validator
from pydantic.dataclasses import (
    dataclass,
    rebuild_dataclass,
)

if TYPE_CHECKING:
    import grpclib.server
    from betterproto.grpc.grpclib_client import MetadataLike
    from grpclib.metadata import Deadline


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallRequest(betterproto.Message):
    """Запрос на передачу события звонка с домофона"""

    to_number: str = betterproto.string_field(1)
    """Вызываемый номер"""

    from_number: str = betterproto.string_field(2)
    """Вызывающий номер"""


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallResponse(betterproto.Message):
    """Ответ на запрос на передачу события звонка с домофона"""

    error: "PostManageCallResponseError | None" = betterproto.message_field(
        1, optional=True, group="type"
    )
    """Ошибка"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallResponseError(betterproto.Message):
    """Ошибка запроса"""

    validation: "ValidationError | None" = betterproto.message_field(
        1, optional=True, group="reason"
    )
    """Ошибка валидации"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallCancelRequest(betterproto.Message):
    """Запрос на передачу события об окончании группового звонка с домофона"""

    to_number: str = betterproto.string_field(1)
    """Вызываемый номер"""

    from_number: str = betterproto.string_field(2)
    """Вызывающий номер"""


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallCancelResponse(betterproto.Message):
    """
    Ответ на запрос на передачу события об окончании группового звонка с домофона
    """

    error: "PostManageCallCancelResponseError | None" = betterproto.message_field(
        1, optional=True, group="type"
    )
    """Ошибка"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PostManageCallCancelResponseError(betterproto.Message):
    """Ошибка запроса"""

    validation: "ValidationError | None" = betterproto.message_field(
        1, optional=True, group="reason"
    )
    """Ошибка валидации"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PutManageUserScopeDetachRequest(betterproto.Message):
    """Запрос метода разъединения аккаунтов"""

    user_id: int = betterproto.int32_field(1)
    """Идентификатор пользователя"""

    scope: str = betterproto.string_field(2)
    """Идентификатор клиента - оператора IoT"""


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PutManageUserScopeDetachResponse(betterproto.Message):
    """Ответ на запрос метода разъединения аккаунтов"""

    error: "PutManageUserScopeDetachResponseError | None" = betterproto.message_field(
        1, optional=True, group="type"
    )
    """Ошибка"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class PutManageUserScopeDetachResponseError(betterproto.Message):
    """Ошибка запроса"""

    validation: "ValidationError | None" = betterproto.message_field(
        1, optional=True, group="reason"
    )
    """Ошибка валидации"""

    @model_validator(mode="after")
    def check_oneof(cls, values):
        return cls._validate_field_groups(values)


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class ValidationError(betterproto.Message):
    """Ошибки валидации"""

    path: str = betterproto.string_field(1)
    """Путь к полю в формате наименования protobuf"""

    message: str = betterproto.string_field(2)
    """Валидационное сообщение"""


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class GetPingRequest(betterproto.Message):
    """Запрос доступности сервиса"""

    pass


@dataclass(eq=False, repr=False, config={"extra": "forbid"})
class GetPingResponse(betterproto.Message):
    """Ответ на запрос доступности сервиса"""

    pass


class IotConnectManageServiceStub(betterproto.ServiceStub):
    async def put_manage_user_scope_detach(
        self,
        put_manage_user_scope_detach_request: "PutManageUserScopeDetachRequest",
        *,
        timeout: "float | None" = None,
        deadline: "Deadline | None" = None,
        metadata: "MetadataLike | None" = None
    ) -> "PutManageUserScopeDetachResponse":
        return await self._unary_unary(
            "/keyapis.iot_connect.v1.IotConnectManageService/PutManageUserScopeDetach",
            put_manage_user_scope_detach_request,
            PutManageUserScopeDetachResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_manage_call(
        self,
        post_manage_call_request: "PostManageCallRequest",
        *,
        timeout: "float | None" = None,
        deadline: "Deadline | None" = None,
        metadata: "MetadataLike | None" = None
    ) -> "PostManageCallResponse":
        return await self._unary_unary(
            "/keyapis.iot_connect.v1.IotConnectManageService/PostManageCall",
            post_manage_call_request,
            PostManageCallResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_manage_call_cancel(
        self,
        post_manage_call_cancel_request: "PostManageCallCancelRequest",
        *,
        timeout: "float | None" = None,
        deadline: "Deadline | None" = None,
        metadata: "MetadataLike | None" = None
    ) -> "PostManageCallCancelResponse":
        return await self._unary_unary(
            "/keyapis.iot_connect.v1.IotConnectManageService/PostManageCallCancel",
            post_manage_call_cancel_request,
            PostManageCallCancelResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class SystemServiceStub(betterproto.ServiceStub):
    async def get_ping(
        self,
        get_ping_request: "GetPingRequest",
        *,
        timeout: "float | None" = None,
        deadline: "Deadline | None" = None,
        metadata: "MetadataLike | None" = None
    ) -> "GetPingResponse":
        return await self._unary_unary(
            "/keyapis.iot_connect.v1.SystemService/GetPing",
            get_ping_request,
            GetPingResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class IotConnectManageServiceBase(ServiceBase):

    async def put_manage_user_scope_detach(
        self, put_manage_user_scope_detach_request: "PutManageUserScopeDetachRequest"
    ) -> "PutManageUserScopeDetachResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_manage_call(
        self, post_manage_call_request: "PostManageCallRequest"
    ) -> "PostManageCallResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_manage_call_cancel(
        self, post_manage_call_cancel_request: "PostManageCallCancelRequest"
    ) -> "PostManageCallCancelResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_put_manage_user_scope_detach(
        self,
        stream: "grpclib.server.Stream[PutManageUserScopeDetachRequest, PutManageUserScopeDetachResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.put_manage_user_scope_detach(request)
        await stream.send_message(response)

    async def __rpc_post_manage_call(
        self,
        stream: "grpclib.server.Stream[PostManageCallRequest, PostManageCallResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_manage_call(request)
        await stream.send_message(response)

    async def __rpc_post_manage_call_cancel(
        self,
        stream: "grpclib.server.Stream[PostManageCallCancelRequest, PostManageCallCancelResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_manage_call_cancel(request)
        await stream.send_message(response)

    def __mapping__(self) -> "dict[str, grpclib.const.Handler]":
        return {
            "/keyapis.iot_connect.v1.IotConnectManageService/PutManageUserScopeDetach": grpclib.const.Handler(
                self.__rpc_put_manage_user_scope_detach,
                grpclib.const.Cardinality.UNARY_UNARY,
                PutManageUserScopeDetachRequest,
                PutManageUserScopeDetachResponse,
            ),
            "/keyapis.iot_connect.v1.IotConnectManageService/PostManageCall": grpclib.const.Handler(
                self.__rpc_post_manage_call,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostManageCallRequest,
                PostManageCallResponse,
            ),
            "/keyapis.iot_connect.v1.IotConnectManageService/PostManageCallCancel": grpclib.const.Handler(
                self.__rpc_post_manage_call_cancel,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostManageCallCancelRequest,
                PostManageCallCancelResponse,
            ),
        }


class SystemServiceBase(ServiceBase):

    async def get_ping(self, get_ping_request: "GetPingRequest") -> "GetPingResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_get_ping(
        self, stream: "grpclib.server.Stream[GetPingRequest, GetPingResponse]"
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_ping(request)
        await stream.send_message(response)

    def __mapping__(self) -> "dict[str, grpclib.const.Handler]":
        return {
            "/keyapis.iot_connect.v1.SystemService/GetPing": grpclib.const.Handler(
                self.__rpc_get_ping,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetPingRequest,
                GetPingResponse,
            ),
        }


rebuild_dataclass(PostManageCallResponse)  # type: ignore
rebuild_dataclass(PostManageCallResponseError)  # type: ignore
rebuild_dataclass(PostManageCallCancelResponse)  # type: ignore
rebuild_dataclass(PostManageCallCancelResponseError)  # type: ignore
rebuild_dataclass(PutManageUserScopeDetachResponse)  # type: ignore
rebuild_dataclass(PutManageUserScopeDetachResponseError)  # type: ignore