# Generated from soulfire/plugin_api.proto. DO NOT EDIT.
# Generated by protoc-gen-connectrpc-py v0.11.1 with parameter "protobuf=google".
# ruff: noqa: PGH004
# ruff: noqa
# fmt: off
from __future__ import annotations
from typing import Protocol, TYPE_CHECKING
from connectrpc.client import ConnectClient, ConnectClientSync
from connectrpc.code import Code
from connectrpc.compat import google_protobuf_binary_codec, google_protobuf_codecs
from connectrpc.compression.gzip import GzipCompression
from connectrpc.errors import ConnectError
from connectrpc.method import IdempotencyLevel, MethodInfo
from connectrpc.protocol import ProtocolType
from connectrpc.server import ConnectASGIApplication, ConnectWSGIApplication, Endpoint, EndpointSync
from pyqwest import Client, SyncClient
from .plugin_api_pb2 import GetPluginApiRequest, GetPluginApiResponse, GetPluginDescriptorSetRequest, GetPluginDescriptorSetResponse, ListPluginApisRequest, ListPluginApisResponse, PluginApiEvent, PluginEvent, WatchPluginApisRequest, WatchPluginEventsRequest
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, AsyncIterator, Iterable, Iterator, Mapping
from connectrpc.codec import Codec
from connectrpc.compression import Compression
from connectrpc.interceptor import Interceptor, InterceptorSync
from connectrpc.request import Headers, RequestContext
_DEFAULT_CODECS = google_protobuf_codecs()
_PROTO_BINARY_CODEC = google_protobuf_binary_codec()
_GZIP_COMPRESSION = GzipCompression()
[docs]
class PluginApiService(Protocol):
[docs]
async def list_plugin_apis(self, request: ListPluginApisRequest, ctx: RequestContext[ListPluginApisRequest, ListPluginApisResponse]) -> ListPluginApisResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_plugin_api(self, request: GetPluginApiRequest, ctx: RequestContext[GetPluginApiRequest, GetPluginApiResponse]) -> GetPluginApiResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_plugin_descriptor_set(self, request: GetPluginDescriptorSetRequest, ctx: RequestContext[GetPluginDescriptorSetRequest, GetPluginDescriptorSetResponse]) -> GetPluginDescriptorSetResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_plugin_apis(self, request: WatchPluginApisRequest, ctx: RequestContext[WatchPluginApisRequest, PluginApiEvent]) -> AsyncIterator[PluginApiEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_plugin_events(self, request: WatchPluginEventsRequest, ctx: RequestContext[WatchPluginEventsRequest, PluginEvent]) -> AsyncIterator[PluginEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class PluginApiServiceASGIApplication(ConnectASGIApplication[PluginApiService]):
def __init__(
self,
service: PluginApiService | AsyncGenerator[PluginApiService],
*,
interceptors: Iterable[Interceptor] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
service=service,
endpoints=lambda svc: {
"/soulfire.v1.PluginApiService/ListPluginApis": Endpoint.unary(
method=MethodInfo(
name="ListPluginApis",
service_name="soulfire.v1.PluginApiService",
input=ListPluginApisRequest,
output=ListPluginApisResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.list_plugin_apis,
),
"/soulfire.v1.PluginApiService/GetPluginApi": Endpoint.unary(
method=MethodInfo(
name="GetPluginApi",
service_name="soulfire.v1.PluginApiService",
input=GetPluginApiRequest,
output=GetPluginApiResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_plugin_api,
),
"/soulfire.v1.PluginApiService/GetPluginDescriptorSet": Endpoint.unary(
method=MethodInfo(
name="GetPluginDescriptorSet",
service_name="soulfire.v1.PluginApiService",
input=GetPluginDescriptorSetRequest,
output=GetPluginDescriptorSetResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_plugin_descriptor_set,
),
"/soulfire.v1.PluginApiService/WatchPluginApis": Endpoint.server_stream(
method=MethodInfo(
name="WatchPluginApis",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginApisRequest,
output=PluginApiEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.watch_plugin_apis,
),
"/soulfire.v1.PluginApiService/WatchPluginEvents": Endpoint.server_stream(
method=MethodInfo(
name="WatchPluginEvents",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginEventsRequest,
output=PluginEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.watch_plugin_events,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.PluginApiService"
[docs]
class PluginApiServiceClient(ConnectClient):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[Interceptor] = (),
http_client: Client | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
async def list_plugin_apis(
self,
request: ListPluginApisRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListPluginApisResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ListPluginApis",
service_name="soulfire.v1.PluginApiService",
input=ListPluginApisRequest,
output=ListPluginApisResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_plugin_api(
self,
request: GetPluginApiRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPluginApiResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetPluginApi",
service_name="soulfire.v1.PluginApiService",
input=GetPluginApiRequest,
output=GetPluginApiResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_plugin_descriptor_set(
self,
request: GetPluginDescriptorSetRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPluginDescriptorSetResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetPluginDescriptorSet",
service_name="soulfire.v1.PluginApiService",
input=GetPluginDescriptorSetRequest,
output=GetPluginDescriptorSetResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_plugin_apis(
self,
request: WatchPluginApisRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[PluginApiEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchPluginApis",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginApisRequest,
output=PluginApiEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_plugin_events(
self,
request: WatchPluginEventsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[PluginEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchPluginEvents",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginEventsRequest,
output=PluginEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class PluginApiServiceSync(Protocol):
[docs]
def list_plugin_apis(self, request: ListPluginApisRequest, ctx: RequestContext[ListPluginApisRequest, ListPluginApisResponse]) -> ListPluginApisResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_plugin_api(self, request: GetPluginApiRequest, ctx: RequestContext[GetPluginApiRequest, GetPluginApiResponse]) -> GetPluginApiResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_plugin_descriptor_set(self, request: GetPluginDescriptorSetRequest, ctx: RequestContext[GetPluginDescriptorSetRequest, GetPluginDescriptorSetResponse]) -> GetPluginDescriptorSetResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_plugin_apis(self, request: WatchPluginApisRequest, ctx: RequestContext[WatchPluginApisRequest, PluginApiEvent]) -> Iterator[PluginApiEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_plugin_events(self, request: WatchPluginEventsRequest, ctx: RequestContext[WatchPluginEventsRequest, PluginEvent]) -> Iterator[PluginEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class PluginApiServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: PluginApiServiceSync,
interceptors: Iterable[InterceptorSync] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
endpoints={
"/soulfire.v1.PluginApiService/ListPluginApis": EndpointSync.unary(
method=MethodInfo(
name="ListPluginApis",
service_name="soulfire.v1.PluginApiService",
input=ListPluginApisRequest,
output=ListPluginApisResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.list_plugin_apis,
),
"/soulfire.v1.PluginApiService/GetPluginApi": EndpointSync.unary(
method=MethodInfo(
name="GetPluginApi",
service_name="soulfire.v1.PluginApiService",
input=GetPluginApiRequest,
output=GetPluginApiResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_plugin_api,
),
"/soulfire.v1.PluginApiService/GetPluginDescriptorSet": EndpointSync.unary(
method=MethodInfo(
name="GetPluginDescriptorSet",
service_name="soulfire.v1.PluginApiService",
input=GetPluginDescriptorSetRequest,
output=GetPluginDescriptorSetResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_plugin_descriptor_set,
),
"/soulfire.v1.PluginApiService/WatchPluginApis": EndpointSync.server_stream(
method=MethodInfo(
name="WatchPluginApis",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginApisRequest,
output=PluginApiEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.watch_plugin_apis,
),
"/soulfire.v1.PluginApiService/WatchPluginEvents": EndpointSync.server_stream(
method=MethodInfo(
name="WatchPluginEvents",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginEventsRequest,
output=PluginEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.watch_plugin_events,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.PluginApiService"
[docs]
class PluginApiServiceClientSync(ConnectClientSync):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[InterceptorSync] = (),
http_client: SyncClient | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
def list_plugin_apis(
self,
request: ListPluginApisRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListPluginApisResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ListPluginApis",
service_name="soulfire.v1.PluginApiService",
input=ListPluginApisRequest,
output=ListPluginApisResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_plugin_api(
self,
request: GetPluginApiRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPluginApiResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetPluginApi",
service_name="soulfire.v1.PluginApiService",
input=GetPluginApiRequest,
output=GetPluginApiResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_plugin_descriptor_set(
self,
request: GetPluginDescriptorSetRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPluginDescriptorSetResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetPluginDescriptorSet",
service_name="soulfire.v1.PluginApiService",
input=GetPluginDescriptorSetRequest,
output=GetPluginDescriptorSetResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_plugin_apis(
self,
request: WatchPluginApisRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[PluginApiEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchPluginApis",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginApisRequest,
output=PluginApiEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_plugin_events(
self,
request: WatchPluginEventsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[PluginEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchPluginEvents",
service_name="soulfire.v1.PluginApiService",
input=WatchPluginEventsRequest,
output=PluginEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)