hermes-chatto-plugin/vendor/chattolib/_pb/chatto/api/v1/threads_connect.py

# -*- coding: utf-8 -*-
# Generated by https://github.com/connectrpc/connect-python.  DO NOT EDIT!
# source: chatto/api/v1/threads.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.api.v1.read_state_pb2 as chatto_dot_api_dot_v1_dot_read__state__pb2
import chatto.api.v1.room_timeline_pb2 as chatto_dot_api_dot_v1_dot_room__timeline__pb2
import chatto.api.v1.threads_pb2 as chatto_dot_api_dot_v1_dot_threads__pb2


class ThreadService(Protocol):
    async def list_followed_threads(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def follow_thread(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def unfollow_thread(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def get_thread_events(self, request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def get_thread_events_around(self, request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")

    async def mark_thread_as_read(self, request: chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class ThreadServiceASGIApplication(ConnectASGIApplication[ThreadService]):
    def __init__(self, service: ThreadService | AsyncGenerator[ThreadService], *, interceptors: Iterable[Interceptor]=(), read_max_bytes: int | None = None, compressions: Iterable[Compression] | None = None) -> None:
        super().__init__(
            service=service,
            endpoints=lambda svc: {
                "/chatto.api.v1.ThreadService/ListFollowedThreads": Endpoint.unary(
                    method=MethodInfo(
                        name="ListFollowedThreads",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.list_followed_threads,
                ),
                "/chatto.api.v1.ThreadService/FollowThread": Endpoint.unary(
                    method=MethodInfo(
                        name="FollowThread",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.follow_thread,
                ),
                "/chatto.api.v1.ThreadService/UnfollowThread": Endpoint.unary(
                    method=MethodInfo(
                        name="UnfollowThread",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.unfollow_thread,
                ),
                "/chatto.api.v1.ThreadService/GetThreadEvents": Endpoint.unary(
                    method=MethodInfo(
                        name="GetThreadEvents",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
                        output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.get_thread_events,
                ),
                "/chatto.api.v1.ThreadService/GetThreadEventsAround": Endpoint.unary(
                    method=MethodInfo(
                        name="GetThreadEventsAround",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
                        output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.get_thread_events_around,
                ),
                "/chatto.api.v1.ThreadService/MarkThreadAsRead": Endpoint.unary(
                    method=MethodInfo(
                        name="MarkThreadAsRead",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
                        output=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=svc.mark_thread_as_read,
                ),
            },
            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.api.v1.ThreadService"


class ThreadServiceClient(ConnectClient):
    async def list_followed_threads(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ListFollowedThreads",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def follow_thread(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="FollowThread",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def unfollow_thread(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="UnfollowThread",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def get_thread_events(
        self,
        request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetThreadEvents",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
                output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def get_thread_events_around(
        self,
        request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetThreadEventsAround",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
                output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    async def mark_thread_as_read(
        self,
        request: chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse:
        return await self.execute_unary(
            request=request,
            method=MethodInfo(
                name="MarkThreadAsRead",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
                output=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )


class ThreadServiceSync(Protocol):
    def list_followed_threads(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def follow_thread(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def unfollow_thread(self, request: chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def get_thread_events(self, request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def get_thread_events_around(self, request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")
    def mark_thread_as_read(self, request: chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest, ctx: RequestContext) -> chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse:
        raise ConnectError(Code.UNIMPLEMENTED, "Not implemented")


class ThreadServiceWSGIApplication(ConnectWSGIApplication):
    def __init__(self, service: ThreadServiceSync, interceptors: Iterable[InterceptorSync]=(), read_max_bytes: int | None = None, compressions: Iterable[Compression] | None = None) -> None:
        super().__init__(
            endpoints={
                "/chatto.api.v1.ThreadService/ListFollowedThreads": EndpointSync.unary(
                    method=MethodInfo(
                        name="ListFollowedThreads",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.list_followed_threads,
                ),
                "/chatto.api.v1.ThreadService/FollowThread": EndpointSync.unary(
                    method=MethodInfo(
                        name="FollowThread",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.follow_thread,
                ),
                "/chatto.api.v1.ThreadService/UnfollowThread": EndpointSync.unary(
                    method=MethodInfo(
                        name="UnfollowThread",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
                        output=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.unfollow_thread,
                ),
                "/chatto.api.v1.ThreadService/GetThreadEvents": EndpointSync.unary(
                    method=MethodInfo(
                        name="GetThreadEvents",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
                        output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.get_thread_events,
                ),
                "/chatto.api.v1.ThreadService/GetThreadEventsAround": EndpointSync.unary(
                    method=MethodInfo(
                        name="GetThreadEventsAround",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
                        output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.get_thread_events_around,
                ),
                "/chatto.api.v1.ThreadService/MarkThreadAsRead": EndpointSync.unary(
                    method=MethodInfo(
                        name="MarkThreadAsRead",
                        service_name="chatto.api.v1.ThreadService",
                        input=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
                        output=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse,
                        idempotency_level=IdempotencyLevel.UNKNOWN,
                    ),
                    function=service.mark_thread_as_read,
                ),
            },
            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.api.v1.ThreadService"


class ThreadServiceClientSync(ConnectClientSync):
    def list_followed_threads(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="ListFollowedThreads",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.ListFollowedThreadsResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def follow_thread(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="FollowThread",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.FollowThreadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def unfollow_thread(
        self,
        request: chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="UnfollowThread",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadRequest,
                output=chatto_dot_api_dot_v1_dot_threads__pb2.UnfollowThreadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def get_thread_events(
        self,
        request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetThreadEvents",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsRequest,
                output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def get_thread_events_around(
        self,
        request: chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="GetThreadEventsAround",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundRequest,
                output=chatto_dot_api_dot_v1_dot_room__timeline__pb2.GetThreadEventsAroundResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )

    def mark_thread_as_read(
        self,
        request: chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
        *,
        headers: Headers | Mapping[str, str] | None = None,
        timeout_ms: int | None = None,
    ) -> chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse:
        return self.execute_unary(
            request=request,
            method=MethodInfo(
                name="MarkThreadAsRead",
                service_name="chatto.api.v1.ThreadService",
                input=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadRequest,
                output=chatto_dot_api_dot_v1_dot_read__state__pb2.MarkThreadAsReadResponse,
                idempotency_level=IdempotencyLevel.UNKNOWN,
            ),
            headers=headers,
            timeout_ms=timeout_ms,
        )