# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: keyapis/identity_b2b/v1/keyapis_identity_b2b_authorization_v1.proto, keyapis/identity_b2b/v1/keyapis_identity_b2b_system_v1.proto
# plugin: python-betterproto
from dataclasses import dataclass
from datetime import datetime
from typing import (
    TYPE_CHECKING,
    Dict,
    Optional,
)

import betterproto
import grpclib
from betterproto.grpc.grpclib_server import ServiceBase

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


@dataclass(eq=False, repr=False)
class GetSystemStatusRequest(betterproto.Message):
    """Запрос проверки доступности сервиса"""

    pass


@dataclass(eq=False, repr=False)
class GetSystemStatusResponse(betterproto.Message):
    """Ответ на запрос проверки доступности сервиса"""

    pass


@dataclass(eq=False, repr=False)
class UserToken(betterproto.Message):
    """Информация о токене пользователя"""

    access_token: str = betterproto.string_field(1)
    """Токен"""

    expired_at: datetime = betterproto.message_field(2)
    """Окончание действия токена"""


@dataclass(eq=False, repr=False)
class Captcha(betterproto.Message):
    """Капча"""

    id: str = betterproto.string_field(1)
    """Идентификатор капчи"""

    url: str = betterproto.string_field(2)
    """Ссылка на капчу"""


@dataclass(eq=False, repr=False)
class CaptchaAnswer(betterproto.Message):
    """Ответ на капчу"""

    id: str = betterproto.string_field(1)
    """Идентификатор капчи"""

    code: str = betterproto.string_field(2)
    """Ответ на капчу"""


@dataclass(eq=False, repr=False)
class CaptchaAnswerError(betterproto.Message):
    """Ошибка ответа на капчу"""

    captcha: "Captcha" = betterproto.message_field(1)
    """Капча"""

    wrong_answer: "CaptchaAnswerErrorWrongAnswer" = betterproto.message_field(
        2, group="reason"
    )
    """Некорректный ответ на капчу"""


@dataclass(eq=False, repr=False)
class CaptchaAnswerErrorWrongAnswer(betterproto.Message):
    """Некорректный ответ на капчу"""

    pass


@dataclass(eq=False, repr=False)
class OtpCode(betterproto.Message):
    """Информация о вводе ОТП-кода"""

    code_id: str = betterproto.string_field(1)
    """Идентификатор ОТП-кода"""

    timeout: int = betterproto.int32_field(2)
    """Время в секундах, оставшееся до возможности отправки следующего кода"""


@dataclass(eq=False, repr=False)
class OtpCodeError(betterproto.Message):
    """Ошибка ответа на капчу"""

    life_time_expired: "OtpCodeErrorLifeTimeExpired" = betterproto.message_field(
        1, group="reason"
    )
    """Код истек"""

    no_attempts: "OtpCodeErrorNoAttempts" = betterproto.message_field(2, group="reason")
    """Попыток больше нет"""

    address_not_found: "OtpCodeErrorAddressNotFound" = betterproto.message_field(
        3, group="reason"
    )
    """
    Адрес для отправки кода не найден. В текущей реализации, эту ошибку
    невозможно получить
    """

    invalid_code: "OtpCodeErrorInvalidCode" = betterproto.message_field(
        4, group="reason"
    )
    """Неправильный ОТП-код"""


@dataclass(eq=False, repr=False)
class OtpCodeErrorLifeTimeExpired(betterproto.Message):
    """Код истек"""

    pass


@dataclass(eq=False, repr=False)
class OtpCodeErrorNoAttempts(betterproto.Message):
    """Попыток больше нет"""

    pass


@dataclass(eq=False, repr=False)
class OtpCodeErrorAddressNotFound(betterproto.Message):
    """Адрес для отправки кода не найден"""

    pass


@dataclass(eq=False, repr=False)
class OtpCodeErrorInvalidCode(betterproto.Message):
    """Неправильный ОТП-код"""

    pass


@dataclass(eq=False, repr=False)
class SsoError(betterproto.Message):
    """Ошибки из ССО"""

    user_is_temporary_blocked: "SsoErrorUserIsTemporaryBlocked" = (
        betterproto.message_field(1, group="reason")
    )
    """Пользователь временно заблокирован"""

    user_not_found: "SsoErrorUserNotFound" = betterproto.message_field(
        2, group="reason"
    )
    """Пользователь не найден"""

    wrong_credentials: "SsoErrorWrongCredentials" = betterproto.message_field(
        3, group="reason"
    )
    """Неверный пароль"""

    user_is_blocked: "SsoErrorUserIsBlocked" = betterproto.message_field(
        4, group="reason"
    )
    """Пользователь заблокирован"""

    interval_exceeded: "SsoErrorIntervalExceeded" = betterproto.message_field(
        5, group="reason"
    )
    """Превышен интервал отправки кода подтверждения"""


@dataclass(eq=False, repr=False)
class SsoErrorUserIsTemporaryBlocked(betterproto.Message):
    """Пользователь временно заблокирован"""

    pass


@dataclass(eq=False, repr=False)
class SsoErrorUserNotFound(betterproto.Message):
    """Пользователь не найден"""

    pass


@dataclass(eq=False, repr=False)
class SsoErrorWrongCredentials(betterproto.Message):
    """Неверный пароль"""

    captcha: "Captcha" = betterproto.message_field(1)
    """Капча"""


@dataclass(eq=False, repr=False)
class SsoErrorUserIsBlocked(betterproto.Message):
    """Пользователь заблокирован"""

    pass


@dataclass(eq=False, repr=False)
class SsoErrorIntervalExceeded(betterproto.Message):
    """Превышен интервал отправки кода подтверждения"""

    pass


@dataclass(eq=False, repr=False)
class KeyError(betterproto.Message):
    """Ошибки сервисов ключа"""

    user_creation: "KeyErrorUserCreation" = betterproto.message_field(1, group="reason")
    """Ошибка создания пользователя"""


@dataclass(eq=False, repr=False)
class KeyErrorUserCreation(betterproto.Message):
    """Ошибка создания пользователя"""

    pass


@dataclass(eq=False, repr=False)
class PostAuthorizationSendCodeRequest(betterproto.Message):
    """Запрос на отправку ОТП-кода для B2B"""

    captcha_answer: "CaptchaAnswer" = betterproto.message_field(1)
    """Ответ на капчу"""

    phone_number: str = betterproto.string_field(2, group="otp_text")
    """Номер телефона"""

    email: str = betterproto.string_field(3, group="otp_text")
    """Электронная почта"""


@dataclass(eq=False, repr=False)
class PostAuthorizationSendCodeResponse(betterproto.Message):
    """Ответ на запрос на отправку ОТП-кода для B2B"""

    error: "PostAuthorizationSendCodeResponseError" = betterproto.message_field(
        1, group="type"
    )
    """Ошибка"""

    data: "OtpCode" = betterproto.message_field(2, group="type")
    """Информация об ОТП-коде"""


@dataclass(eq=False, repr=False)
class PostAuthorizationSendCodeResponseError(betterproto.Message):
    """Ошибка запроса на отправку ОТП-кода"""

    captcha_answer: "CaptchaAnswerError" = betterproto.message_field(1, group="reason")
    """Ошибка ответа с капчей"""

    sso: "SsoError" = betterproto.message_field(2, group="reason")
    """Ошибка из ССО"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginRequest(betterproto.Message):
    """Запрос на авторизацию по ОТП-коду для B2B"""

    code_id: str = betterproto.string_field(1)
    """Идентификатор ОТП-кода"""

    code: str = betterproto.string_field(2)
    """Значение ОТП-кода"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginResponse(betterproto.Message):
    """Ответ на запрос об авторизации пользователя B2B"""

    error: "PostAuthorizationLoginResponseError" = betterproto.message_field(
        1, group="type"
    )
    """Ошибка"""

    data: "UserToken" = betterproto.message_field(2, group="type")
    """Токен пользователя"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginResponseError(betterproto.Message):
    """Ошибка запроса авторизации пользователя"""

    otp_code: "OtpCodeError" = betterproto.message_field(1, group="reason")
    """Ошибка отп-кода"""

    sso: "SsoError" = betterproto.message_field(2, group="reason")
    """Ошибка из ССО"""

    key: "KeyError" = betterproto.message_field(3, group="reason")
    """Ошибка из сервисов ключа"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginByPasswordRequest(betterproto.Message):
    """Запрос на авторизацию по паролю для B2B"""

    password: str = betterproto.string_field(1)
    """Пароль"""

    captcha_answer: "CaptchaAnswer" = betterproto.message_field(2)
    """Ответ на капчу"""

    phone_number: str = betterproto.string_field(3, group="type")
    """Номер телефона"""

    email: str = betterproto.string_field(4, group="type")
    """Электронная почта"""

    login: str = betterproto.string_field(5, group="type")
    """Логин"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginByPasswordResponse(betterproto.Message):
    """Ответ на запрос об авторизации по паролю пользователя B2B"""

    error: "PostAuthorizationLoginByPasswordResponseError" = betterproto.message_field(
        1, group="type"
    )
    """Ошибка"""

    data: "UserToken" = betterproto.message_field(2, group="type")
    """Токен пользователя"""

    data2_fa: "Data2Fa" = betterproto.message_field(3, group="type")
    """Данные для двухфакторной аутентификации"""


@dataclass(eq=False, repr=False)
class PostAuthorizationLoginByPasswordResponseError(betterproto.Message):
    """Ошибка запроса авторизации по паролю пользователя"""

    captcha_answer: "CaptchaAnswerError" = betterproto.message_field(1, group="reason")
    """Ошибка капчи"""

    sso: "SsoError" = betterproto.message_field(2, group="reason")
    """Ошибка из ССО"""

    key: "KeyError" = betterproto.message_field(3, group="reason")
    """Ошибка из сервисов ключа"""

    otp_code: "OtpCodeError" = betterproto.message_field(4, group="reason")
    """Ошибка отп-кода"""


@dataclass(eq=False, repr=False)
class PostAuthorizationCheckCodeRequest(betterproto.Message):
    """
    Запрос проверки кода двухфакторной авторизации и активация сессии для B2B
    """

    session_id: str = betterproto.string_field(1)
    """Идентификатор сессии двухфакторной аутентификации"""

    action_id: str = betterproto.string_field(2)
    """Идентификатор процесса двухфакторной аутентификации"""

    code_id: str = betterproto.string_field(3)
    """Идентификатор кода двухфакторной авторизации"""

    code: str = betterproto.string_field(4)
    """Код, введенный пользователем"""


@dataclass(eq=False, repr=False)
class PostAuthorizationCheckCodeResponse(betterproto.Message):
    """
    Ответ на запрос проверки кода двухфакторной авторизации и активация сессии
    для B2B
    """

    error: "PostAuthorizationCheckCodeResponseError" = betterproto.message_field(
        1, group="type"
    )
    """Ошибка"""

    data: "UserToken" = betterproto.message_field(2, group="type")
    """Токен пользователя"""


@dataclass(eq=False, repr=False)
class PostAuthorizationCheckCodeResponseError(betterproto.Message):
    """Ошибка запроса авторизации по паролю пользователя"""

    sso: "SsoError" = betterproto.message_field(1, group="reason")
    """Ошибка из ССО"""

    key: "KeyError" = betterproto.message_field(2, group="reason")
    """Ошибка из сервисов ключа"""

    otp_code: "OtpCodeError" = betterproto.message_field(3, group="reason")
    """Ошибка кода подтверждения"""


@dataclass(eq=False, repr=False)
class Data2Fa(betterproto.Message):
    """Данные для двухфакторной аутентификации"""

    session_id: str = betterproto.string_field(1)
    """Идентификатор сессии двухфакторной аутентификации"""

    action_id: str = betterproto.string_field(2)
    """Идентификатор процесса двухфакторной аутентификации"""

    code_id: str = betterproto.string_field(3)
    """Идентификатор кода двухфакторной авторизации"""

    timeout: int = betterproto.int32_field(4)
    """Время в секундах, оставшееся до возможности отправки следующего кода"""


class SystemServiceStub(betterproto.ServiceStub):
    async def get_system_status(
        self,
        get_system_status_request: "GetSystemStatusRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "GetSystemStatusResponse":
        return await self._unary_unary(
            "/keyapis.identity_b2b.v1.SystemService/GetSystemStatus",
            get_system_status_request,
            GetSystemStatusResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class AuthorizationB2BServiceStub(betterproto.ServiceStub):
    async def post_authorization_login(
        self,
        post_authorization_login_request: "PostAuthorizationLoginRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostAuthorizationLoginResponse":
        return await self._unary_unary(
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationLogin",
            post_authorization_login_request,
            PostAuthorizationLoginResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_authorization_send_code(
        self,
        post_authorization_send_code_request: "PostAuthorizationSendCodeRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostAuthorizationSendCodeResponse":
        return await self._unary_unary(
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationSendCode",
            post_authorization_send_code_request,
            PostAuthorizationSendCodeResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_authorization_login_by_password(
        self,
        post_authorization_login_by_password_request: "PostAuthorizationLoginByPasswordRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostAuthorizationLoginByPasswordResponse":
        return await self._unary_unary(
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationLoginByPassword",
            post_authorization_login_by_password_request,
            PostAuthorizationLoginByPasswordResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )

    async def post_authorization_check_code(
        self,
        post_authorization_check_code_request: "PostAuthorizationCheckCodeRequest",
        *,
        timeout: Optional[float] = None,
        deadline: Optional["Deadline"] = None,
        metadata: Optional["MetadataLike"] = None
    ) -> "PostAuthorizationCheckCodeResponse":
        return await self._unary_unary(
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationCheckCode",
            post_authorization_check_code_request,
            PostAuthorizationCheckCodeResponse,
            timeout=timeout,
            deadline=deadline,
            metadata=metadata,
        )


class SystemServiceBase(ServiceBase):

    async def get_system_status(
        self, get_system_status_request: "GetSystemStatusRequest"
    ) -> "GetSystemStatusResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_get_system_status(
        self,
        stream: "grpclib.server.Stream[GetSystemStatusRequest, GetSystemStatusResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.get_system_status(request)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/keyapis.identity_b2b.v1.SystemService/GetSystemStatus": grpclib.const.Handler(
                self.__rpc_get_system_status,
                grpclib.const.Cardinality.UNARY_UNARY,
                GetSystemStatusRequest,
                GetSystemStatusResponse,
            ),
        }


class AuthorizationB2BServiceBase(ServiceBase):

    async def post_authorization_login(
        self, post_authorization_login_request: "PostAuthorizationLoginRequest"
    ) -> "PostAuthorizationLoginResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_authorization_send_code(
        self, post_authorization_send_code_request: "PostAuthorizationSendCodeRequest"
    ) -> "PostAuthorizationSendCodeResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_authorization_login_by_password(
        self,
        post_authorization_login_by_password_request: "PostAuthorizationLoginByPasswordRequest",
    ) -> "PostAuthorizationLoginByPasswordResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def post_authorization_check_code(
        self, post_authorization_check_code_request: "PostAuthorizationCheckCodeRequest"
    ) -> "PostAuthorizationCheckCodeResponse":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_post_authorization_login(
        self,
        stream: "grpclib.server.Stream[PostAuthorizationLoginRequest, PostAuthorizationLoginResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_authorization_login(request)
        await stream.send_message(response)

    async def __rpc_post_authorization_send_code(
        self,
        stream: "grpclib.server.Stream[PostAuthorizationSendCodeRequest, PostAuthorizationSendCodeResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_authorization_send_code(request)
        await stream.send_message(response)

    async def __rpc_post_authorization_login_by_password(
        self,
        stream: "grpclib.server.Stream[PostAuthorizationLoginByPasswordRequest, PostAuthorizationLoginByPasswordResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_authorization_login_by_password(request)
        await stream.send_message(response)

    async def __rpc_post_authorization_check_code(
        self,
        stream: "grpclib.server.Stream[PostAuthorizationCheckCodeRequest, PostAuthorizationCheckCodeResponse]",
    ) -> None:
        request = await stream.recv_message()
        response = await self.post_authorization_check_code(request)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationLogin": grpclib.const.Handler(
                self.__rpc_post_authorization_login,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostAuthorizationLoginRequest,
                PostAuthorizationLoginResponse,
            ),
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationSendCode": grpclib.const.Handler(
                self.__rpc_post_authorization_send_code,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostAuthorizationSendCodeRequest,
                PostAuthorizationSendCodeResponse,
            ),
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationLoginByPassword": grpclib.const.Handler(
                self.__rpc_post_authorization_login_by_password,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostAuthorizationLoginByPasswordRequest,
                PostAuthorizationLoginByPasswordResponse,
            ),
            "/keyapis.identity_b2b.v1.AuthorizationB2bService/PostAuthorizationCheckCode": grpclib.const.Handler(
                self.__rpc_post_authorization_check_code,
                grpclib.const.Cardinality.UNARY_UNARY,
                PostAuthorizationCheckCodeRequest,
                PostAuthorizationCheckCodeResponse,
            ),
        }