# Generated from soulfire/client.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 .client_pb2 import ClientDataRequest, ClientDataResponse, GenerateAPITokenRequest, GenerateAPITokenResponse, GenerateWebDAVTokenRequest, GenerateWebDAVTokenResponse, InvalidateSelfSessionsRequest, InvalidateSelfSessionsResponse, UpdateSelfEmailRequest, UpdateSelfEmailResponse, UpdateSelfUsernameRequest, UpdateSelfUsernameResponse
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Iterable, 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 ClientService(Protocol):
[docs]
async def get_client_data(self, request: ClientDataRequest, ctx: RequestContext[ClientDataRequest, ClientDataResponse]) -> ClientDataResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def generate_web_dav_token(self, request: GenerateWebDAVTokenRequest, ctx: RequestContext[GenerateWebDAVTokenRequest, GenerateWebDAVTokenResponse]) -> GenerateWebDAVTokenResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def generate_api_token(self, request: GenerateAPITokenRequest, ctx: RequestContext[GenerateAPITokenRequest, GenerateAPITokenResponse]) -> GenerateAPITokenResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def update_self_username(self, request: UpdateSelfUsernameRequest, ctx: RequestContext[UpdateSelfUsernameRequest, UpdateSelfUsernameResponse]) -> UpdateSelfUsernameResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def update_self_email(self, request: UpdateSelfEmailRequest, ctx: RequestContext[UpdateSelfEmailRequest, UpdateSelfEmailResponse]) -> UpdateSelfEmailResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def invalidate_self_sessions(self, request: InvalidateSelfSessionsRequest, ctx: RequestContext[InvalidateSelfSessionsRequest, InvalidateSelfSessionsResponse]) -> InvalidateSelfSessionsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class ClientServiceASGIApplication(ConnectASGIApplication[ClientService]):
def __init__(
self,
service: ClientService | AsyncGenerator[ClientService],
*,
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.ClientService/GetClientData": Endpoint.unary(
method=MethodInfo(
name="GetClientData",
service_name="soulfire.v1.ClientService",
input=ClientDataRequest,
output=ClientDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_client_data,
),
"/soulfire.v1.ClientService/GenerateWebDAVToken": Endpoint.unary(
method=MethodInfo(
name="GenerateWebDAVToken",
service_name="soulfire.v1.ClientService",
input=GenerateWebDAVTokenRequest,
output=GenerateWebDAVTokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.generate_web_dav_token,
),
"/soulfire.v1.ClientService/GenerateAPIToken": Endpoint.unary(
method=MethodInfo(
name="GenerateAPIToken",
service_name="soulfire.v1.ClientService",
input=GenerateAPITokenRequest,
output=GenerateAPITokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.generate_api_token,
),
"/soulfire.v1.ClientService/UpdateSelfUsername": Endpoint.unary(
method=MethodInfo(
name="UpdateSelfUsername",
service_name="soulfire.v1.ClientService",
input=UpdateSelfUsernameRequest,
output=UpdateSelfUsernameResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.update_self_username,
),
"/soulfire.v1.ClientService/UpdateSelfEmail": Endpoint.unary(
method=MethodInfo(
name="UpdateSelfEmail",
service_name="soulfire.v1.ClientService",
input=UpdateSelfEmailRequest,
output=UpdateSelfEmailResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.update_self_email,
),
"/soulfire.v1.ClientService/InvalidateSelfSessions": Endpoint.unary(
method=MethodInfo(
name="InvalidateSelfSessions",
service_name="soulfire.v1.ClientService",
input=InvalidateSelfSessionsRequest,
output=InvalidateSelfSessionsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.invalidate_self_sessions,
),
},
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.ClientService"
[docs]
class ClientServiceClient(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 get_client_data(
self,
request: ClientDataRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ClientDataResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetClientData",
service_name="soulfire.v1.ClientService",
input=ClientDataRequest,
output=ClientDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def generate_web_dav_token(
self,
request: GenerateWebDAVTokenRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GenerateWebDAVTokenResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GenerateWebDAVToken",
service_name="soulfire.v1.ClientService",
input=GenerateWebDAVTokenRequest,
output=GenerateWebDAVTokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def generate_api_token(
self,
request: GenerateAPITokenRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GenerateAPITokenResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GenerateAPIToken",
service_name="soulfire.v1.ClientService",
input=GenerateAPITokenRequest,
output=GenerateAPITokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def update_self_username(
self,
request: UpdateSelfUsernameRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSelfUsernameResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSelfUsername",
service_name="soulfire.v1.ClientService",
input=UpdateSelfUsernameRequest,
output=UpdateSelfUsernameResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def update_self_email(
self,
request: UpdateSelfEmailRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSelfEmailResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSelfEmail",
service_name="soulfire.v1.ClientService",
input=UpdateSelfEmailRequest,
output=UpdateSelfEmailResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def invalidate_self_sessions(
self,
request: InvalidateSelfSessionsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InvalidateSelfSessionsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="InvalidateSelfSessions",
service_name="soulfire.v1.ClientService",
input=InvalidateSelfSessionsRequest,
output=InvalidateSelfSessionsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class ClientServiceSync(Protocol):
[docs]
def get_client_data(self, request: ClientDataRequest, ctx: RequestContext[ClientDataRequest, ClientDataResponse]) -> ClientDataResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def generate_web_dav_token(self, request: GenerateWebDAVTokenRequest, ctx: RequestContext[GenerateWebDAVTokenRequest, GenerateWebDAVTokenResponse]) -> GenerateWebDAVTokenResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def generate_api_token(self, request: GenerateAPITokenRequest, ctx: RequestContext[GenerateAPITokenRequest, GenerateAPITokenResponse]) -> GenerateAPITokenResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def update_self_username(self, request: UpdateSelfUsernameRequest, ctx: RequestContext[UpdateSelfUsernameRequest, UpdateSelfUsernameResponse]) -> UpdateSelfUsernameResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def update_self_email(self, request: UpdateSelfEmailRequest, ctx: RequestContext[UpdateSelfEmailRequest, UpdateSelfEmailResponse]) -> UpdateSelfEmailResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def invalidate_self_sessions(self, request: InvalidateSelfSessionsRequest, ctx: RequestContext[InvalidateSelfSessionsRequest, InvalidateSelfSessionsResponse]) -> InvalidateSelfSessionsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class ClientServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: ClientServiceSync,
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.ClientService/GetClientData": EndpointSync.unary(
method=MethodInfo(
name="GetClientData",
service_name="soulfire.v1.ClientService",
input=ClientDataRequest,
output=ClientDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_client_data,
),
"/soulfire.v1.ClientService/GenerateWebDAVToken": EndpointSync.unary(
method=MethodInfo(
name="GenerateWebDAVToken",
service_name="soulfire.v1.ClientService",
input=GenerateWebDAVTokenRequest,
output=GenerateWebDAVTokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.generate_web_dav_token,
),
"/soulfire.v1.ClientService/GenerateAPIToken": EndpointSync.unary(
method=MethodInfo(
name="GenerateAPIToken",
service_name="soulfire.v1.ClientService",
input=GenerateAPITokenRequest,
output=GenerateAPITokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.generate_api_token,
),
"/soulfire.v1.ClientService/UpdateSelfUsername": EndpointSync.unary(
method=MethodInfo(
name="UpdateSelfUsername",
service_name="soulfire.v1.ClientService",
input=UpdateSelfUsernameRequest,
output=UpdateSelfUsernameResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.update_self_username,
),
"/soulfire.v1.ClientService/UpdateSelfEmail": EndpointSync.unary(
method=MethodInfo(
name="UpdateSelfEmail",
service_name="soulfire.v1.ClientService",
input=UpdateSelfEmailRequest,
output=UpdateSelfEmailResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.update_self_email,
),
"/soulfire.v1.ClientService/InvalidateSelfSessions": EndpointSync.unary(
method=MethodInfo(
name="InvalidateSelfSessions",
service_name="soulfire.v1.ClientService",
input=InvalidateSelfSessionsRequest,
output=InvalidateSelfSessionsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.invalidate_self_sessions,
),
},
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.ClientService"
[docs]
class ClientServiceClientSync(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 get_client_data(
self,
request: ClientDataRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ClientDataResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetClientData",
service_name="soulfire.v1.ClientService",
input=ClientDataRequest,
output=ClientDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def generate_web_dav_token(
self,
request: GenerateWebDAVTokenRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GenerateWebDAVTokenResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GenerateWebDAVToken",
service_name="soulfire.v1.ClientService",
input=GenerateWebDAVTokenRequest,
output=GenerateWebDAVTokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def generate_api_token(
self,
request: GenerateAPITokenRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GenerateAPITokenResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GenerateAPIToken",
service_name="soulfire.v1.ClientService",
input=GenerateAPITokenRequest,
output=GenerateAPITokenResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def update_self_username(
self,
request: UpdateSelfUsernameRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSelfUsernameResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSelfUsername",
service_name="soulfire.v1.ClientService",
input=UpdateSelfUsernameRequest,
output=UpdateSelfUsernameResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def update_self_email(
self,
request: UpdateSelfEmailRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSelfEmailResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSelfEmail",
service_name="soulfire.v1.ClientService",
input=UpdateSelfEmailRequest,
output=UpdateSelfEmailResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def invalidate_self_sessions(
self,
request: InvalidateSelfSessionsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InvalidateSelfSessionsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="InvalidateSelfSessions",
service_name="soulfire.v1.ClientService",
input=InvalidateSelfSessionsRequest,
output=InvalidateSelfSessionsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)