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

# -*- coding: utf-8 -*-
# Generated by https://github.com/connectrpc/connect-python.  DO NOT EDIT!
# source: chatto/admin/v1/roles.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.roles_pb2 as chatto_dot_admin_dot_v1_dot_roles__pb2


class AdminRoleService(Protocol):
    async def list_roles(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def get_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def create_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def update_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def delete_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def reorder_roles(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class AdminRoleServiceASGIApplication(ConnectASGIApplication[AdminRoleService]):
    def __init__(self, service: AdminRoleService | AsyncGenerator[AdminRoleService], *, 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.AdminRoleService/ListRoles": Endpoint.unary(
                    method=MethodInfo(
                        name="ListRoles",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.list_roles,
                ),
                "/chatto.admin.v1.AdminRoleService/GetRole": Endpoint.unary(
                    method=MethodInfo(
                        name="GetRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.get_role,
                ),
                "/chatto.admin.v1.AdminRoleService/CreateRole": Endpoint.unary(
                    method=MethodInfo(
                        name="CreateRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.create_role,
                ),
                "/chatto.admin.v1.AdminRoleService/UpdateRole": Endpoint.unary(
                    method=MethodInfo(
                        name="UpdateRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.update_role,
                ),
                "/chatto.admin.v1.AdminRoleService/DeleteRole": Endpoint.unary(
                    method=MethodInfo(
                        name="DeleteRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.delete_role,
                ),
                "/chatto.admin.v1.AdminRoleService/ReorderRoles": Endpoint.unary(
                    method=MethodInfo(
                        name="ReorderRoles",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.reorder_roles,
                ),
            },
            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.AdminRoleService"


class AdminRoleServiceClient(ConnectClient):
    async def list_roles(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ListRoles",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def get_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def create_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="CreateRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def update_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="UpdateRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def delete_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="DeleteRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def reorder_roles(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ReorderRoles",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )


class AdminRoleServiceSync(Protocol):
    def list_roles(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def get_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def create_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def update_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def delete_role(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def reorder_roles(self, request: chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest, ctx: RequestContext) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class AdminRoleServiceWSGIApplication(ConnectWSGIApplication):
    def __init__(self, service: AdminRoleServiceSync, interceptors: Iterable[InterceptorSync]=(), read_max_bytes: int | None = None, compressions: Iterable[Compression] | None = None) -> None:
        super().__init__(
            endpoints={
                "/chatto.admin.v1.AdminRoleService/ListRoles": EndpointSync.unary(
                    method=MethodInfo(
                        name="ListRoles",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.list_roles,
                ),
                "/chatto.admin.v1.AdminRoleService/GetRole": EndpointSync.unary(
                    method=MethodInfo(
                        name="GetRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.get_role,
                ),
                "/chatto.admin.v1.AdminRoleService/CreateRole": EndpointSync.unary(
                    method=MethodInfo(
                        name="CreateRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.create_role,
                ),
                "/chatto.admin.v1.AdminRoleService/UpdateRole": EndpointSync.unary(
                    method=MethodInfo(
                        name="UpdateRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.update_role,
                ),
                "/chatto.admin.v1.AdminRoleService/DeleteRole": EndpointSync.unary(
                    method=MethodInfo(
                        name="DeleteRole",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.delete_role,
                ),
                "/chatto.admin.v1.AdminRoleService/ReorderRoles": EndpointSync.unary(
                    method=MethodInfo(
                        name="ReorderRoles",
                        service_name="chatto.admin.v1.AdminRoleService",
                        input=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
                        output=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.reorder_roles,
                ),
            },
            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.AdminRoleService"


class AdminRoleServiceClientSync(ConnectClientSync):
    def list_roles(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ListRoles",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.ListRolesResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def get_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.GetRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def create_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="CreateRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.CreateRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def update_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="UpdateRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.UpdateRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def delete_role(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="DeleteRole",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.DeleteRoleResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def reorder_roles(
        self,
        request: chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ReorderRoles",
                service_name="chatto.admin.v1.AdminRoleService",
                input=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesRequest,
                output=chatto_dot_admin_dot_v1_dot_roles__pb2.ReorderRolesResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )