# Generated from soulfire/script.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 .script_pb2 import ActivateScriptRequest, CreateScriptRequest, CreateScriptResponse, DeactivateScriptRequest, DeactivateScriptResponse, DeleteScriptRequest, DeleteScriptResponse, DryRunScriptRequest, GetNodeTypesRequest, GetNodeTypesResponse, GetRegistryDataRequest, GetRegistryDataResponse, GetScriptRequest, GetScriptResponse, GetScriptStatusRequest, GetScriptStatusResponse, ListScriptsRequest, ListScriptsResponse, ScriptEvent, ScriptLogEntry, SubscribeScriptLogsRequest, UpdateScriptRequest, UpdateScriptResponse, ValidateScriptRequest, ValidateScriptResponse
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 ScriptService(Protocol):
[docs]
async def create_script(self, request: CreateScriptRequest, ctx: RequestContext[CreateScriptRequest, CreateScriptResponse]) -> CreateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_script(self, request: GetScriptRequest, ctx: RequestContext[GetScriptRequest, GetScriptResponse]) -> GetScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def update_script(self, request: UpdateScriptRequest, ctx: RequestContext[UpdateScriptRequest, UpdateScriptResponse]) -> UpdateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def delete_script(self, request: DeleteScriptRequest, ctx: RequestContext[DeleteScriptRequest, DeleteScriptResponse]) -> DeleteScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def list_scripts(self, request: ListScriptsRequest, ctx: RequestContext[ListScriptsRequest, ListScriptsResponse]) -> ListScriptsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def activate_script(self, request: ActivateScriptRequest, ctx: RequestContext[ActivateScriptRequest, ScriptEvent]) -> AsyncIterator[ScriptEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def deactivate_script(self, request: DeactivateScriptRequest, ctx: RequestContext[DeactivateScriptRequest, DeactivateScriptResponse]) -> DeactivateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_script_status(self, request: GetScriptStatusRequest, ctx: RequestContext[GetScriptStatusRequest, GetScriptStatusResponse]) -> GetScriptStatusResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def subscribe_script_logs(self, request: SubscribeScriptLogsRequest, ctx: RequestContext[SubscribeScriptLogsRequest, ScriptLogEntry]) -> AsyncIterator[ScriptLogEntry]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_node_types(self, request: GetNodeTypesRequest, ctx: RequestContext[GetNodeTypesRequest, GetNodeTypesResponse]) -> GetNodeTypesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_registry_data(self, request: GetRegistryDataRequest, ctx: RequestContext[GetRegistryDataRequest, GetRegistryDataResponse]) -> GetRegistryDataResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def validate_script(self, request: ValidateScriptRequest, ctx: RequestContext[ValidateScriptRequest, ValidateScriptResponse]) -> ValidateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def dry_run_script(self, request: DryRunScriptRequest, ctx: RequestContext[DryRunScriptRequest, ScriptEvent]) -> AsyncIterator[ScriptEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class ScriptServiceASGIApplication(ConnectASGIApplication[ScriptService]):
def __init__(
self,
service: ScriptService | AsyncGenerator[ScriptService],
*,
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.ScriptService/CreateScript": Endpoint.unary(
method=MethodInfo(
name="CreateScript",
service_name="soulfire.v1.ScriptService",
input=CreateScriptRequest,
output=CreateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.create_script,
),
"/soulfire.v1.ScriptService/GetScript": Endpoint.unary(
method=MethodInfo(
name="GetScript",
service_name="soulfire.v1.ScriptService",
input=GetScriptRequest,
output=GetScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_script,
),
"/soulfire.v1.ScriptService/UpdateScript": Endpoint.unary(
method=MethodInfo(
name="UpdateScript",
service_name="soulfire.v1.ScriptService",
input=UpdateScriptRequest,
output=UpdateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.update_script,
),
"/soulfire.v1.ScriptService/DeleteScript": Endpoint.unary(
method=MethodInfo(
name="DeleteScript",
service_name="soulfire.v1.ScriptService",
input=DeleteScriptRequest,
output=DeleteScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.delete_script,
),
"/soulfire.v1.ScriptService/ListScripts": Endpoint.unary(
method=MethodInfo(
name="ListScripts",
service_name="soulfire.v1.ScriptService",
input=ListScriptsRequest,
output=ListScriptsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.list_scripts,
),
"/soulfire.v1.ScriptService/ActivateScript": Endpoint.server_stream(
method=MethodInfo(
name="ActivateScript",
service_name="soulfire.v1.ScriptService",
input=ActivateScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.activate_script,
),
"/soulfire.v1.ScriptService/DeactivateScript": Endpoint.unary(
method=MethodInfo(
name="DeactivateScript",
service_name="soulfire.v1.ScriptService",
input=DeactivateScriptRequest,
output=DeactivateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.deactivate_script,
),
"/soulfire.v1.ScriptService/GetScriptStatus": Endpoint.unary(
method=MethodInfo(
name="GetScriptStatus",
service_name="soulfire.v1.ScriptService",
input=GetScriptStatusRequest,
output=GetScriptStatusResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_script_status,
),
"/soulfire.v1.ScriptService/SubscribeScriptLogs": Endpoint.server_stream(
method=MethodInfo(
name="SubscribeScriptLogs",
service_name="soulfire.v1.ScriptService",
input=SubscribeScriptLogsRequest,
output=ScriptLogEntry,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.subscribe_script_logs,
),
"/soulfire.v1.ScriptService/GetNodeTypes": Endpoint.unary(
method=MethodInfo(
name="GetNodeTypes",
service_name="soulfire.v1.ScriptService",
input=GetNodeTypesRequest,
output=GetNodeTypesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_node_types,
),
"/soulfire.v1.ScriptService/GetRegistryData": Endpoint.unary(
method=MethodInfo(
name="GetRegistryData",
service_name="soulfire.v1.ScriptService",
input=GetRegistryDataRequest,
output=GetRegistryDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_registry_data,
),
"/soulfire.v1.ScriptService/ValidateScript": Endpoint.unary(
method=MethodInfo(
name="ValidateScript",
service_name="soulfire.v1.ScriptService",
input=ValidateScriptRequest,
output=ValidateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.validate_script,
),
"/soulfire.v1.ScriptService/DryRunScript": Endpoint.server_stream(
method=MethodInfo(
name="DryRunScript",
service_name="soulfire.v1.ScriptService",
input=DryRunScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.dry_run_script,
),
},
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.ScriptService"
[docs]
class ScriptServiceClient(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 create_script(
self,
request: CreateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CreateScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CreateScript",
service_name="soulfire.v1.ScriptService",
input=CreateScriptRequest,
output=CreateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_script(
self,
request: GetScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetScript",
service_name="soulfire.v1.ScriptService",
input=GetScriptRequest,
output=GetScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def update_script(
self,
request: UpdateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateScript",
service_name="soulfire.v1.ScriptService",
input=UpdateScriptRequest,
output=UpdateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def delete_script(
self,
request: DeleteScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DeleteScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="DeleteScript",
service_name="soulfire.v1.ScriptService",
input=DeleteScriptRequest,
output=DeleteScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def list_scripts(
self,
request: ListScriptsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListScriptsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ListScripts",
service_name="soulfire.v1.ScriptService",
input=ListScriptsRequest,
output=ListScriptsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def activate_script(
self,
request: ActivateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[ScriptEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="ActivateScript",
service_name="soulfire.v1.ScriptService",
input=ActivateScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def deactivate_script(
self,
request: DeactivateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DeactivateScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="DeactivateScript",
service_name="soulfire.v1.ScriptService",
input=DeactivateScriptRequest,
output=DeactivateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_script_status(
self,
request: GetScriptStatusRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetScriptStatusResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetScriptStatus",
service_name="soulfire.v1.ScriptService",
input=GetScriptStatusRequest,
output=GetScriptStatusResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def subscribe_script_logs(
self,
request: SubscribeScriptLogsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[ScriptLogEntry]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="SubscribeScriptLogs",
service_name="soulfire.v1.ScriptService",
input=SubscribeScriptLogsRequest,
output=ScriptLogEntry,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_node_types(
self,
request: GetNodeTypesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetNodeTypesResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetNodeTypes",
service_name="soulfire.v1.ScriptService",
input=GetNodeTypesRequest,
output=GetNodeTypesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_registry_data(
self,
request: GetRegistryDataRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetRegistryDataResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetRegistryData",
service_name="soulfire.v1.ScriptService",
input=GetRegistryDataRequest,
output=GetRegistryDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def validate_script(
self,
request: ValidateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ValidateScriptResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ValidateScript",
service_name="soulfire.v1.ScriptService",
input=ValidateScriptRequest,
output=ValidateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def dry_run_script(
self,
request: DryRunScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[ScriptEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="DryRunScript",
service_name="soulfire.v1.ScriptService",
input=DryRunScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class ScriptServiceSync(Protocol):
[docs]
def create_script(self, request: CreateScriptRequest, ctx: RequestContext[CreateScriptRequest, CreateScriptResponse]) -> CreateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_script(self, request: GetScriptRequest, ctx: RequestContext[GetScriptRequest, GetScriptResponse]) -> GetScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def update_script(self, request: UpdateScriptRequest, ctx: RequestContext[UpdateScriptRequest, UpdateScriptResponse]) -> UpdateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def delete_script(self, request: DeleteScriptRequest, ctx: RequestContext[DeleteScriptRequest, DeleteScriptResponse]) -> DeleteScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def list_scripts(self, request: ListScriptsRequest, ctx: RequestContext[ListScriptsRequest, ListScriptsResponse]) -> ListScriptsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def activate_script(self, request: ActivateScriptRequest, ctx: RequestContext[ActivateScriptRequest, ScriptEvent]) -> Iterator[ScriptEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def deactivate_script(self, request: DeactivateScriptRequest, ctx: RequestContext[DeactivateScriptRequest, DeactivateScriptResponse]) -> DeactivateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_script_status(self, request: GetScriptStatusRequest, ctx: RequestContext[GetScriptStatusRequest, GetScriptStatusResponse]) -> GetScriptStatusResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def subscribe_script_logs(self, request: SubscribeScriptLogsRequest, ctx: RequestContext[SubscribeScriptLogsRequest, ScriptLogEntry]) -> Iterator[ScriptLogEntry]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_node_types(self, request: GetNodeTypesRequest, ctx: RequestContext[GetNodeTypesRequest, GetNodeTypesResponse]) -> GetNodeTypesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_registry_data(self, request: GetRegistryDataRequest, ctx: RequestContext[GetRegistryDataRequest, GetRegistryDataResponse]) -> GetRegistryDataResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def validate_script(self, request: ValidateScriptRequest, ctx: RequestContext[ValidateScriptRequest, ValidateScriptResponse]) -> ValidateScriptResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def dry_run_script(self, request: DryRunScriptRequest, ctx: RequestContext[DryRunScriptRequest, ScriptEvent]) -> Iterator[ScriptEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class ScriptServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: ScriptServiceSync,
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.ScriptService/CreateScript": EndpointSync.unary(
method=MethodInfo(
name="CreateScript",
service_name="soulfire.v1.ScriptService",
input=CreateScriptRequest,
output=CreateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.create_script,
),
"/soulfire.v1.ScriptService/GetScript": EndpointSync.unary(
method=MethodInfo(
name="GetScript",
service_name="soulfire.v1.ScriptService",
input=GetScriptRequest,
output=GetScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_script,
),
"/soulfire.v1.ScriptService/UpdateScript": EndpointSync.unary(
method=MethodInfo(
name="UpdateScript",
service_name="soulfire.v1.ScriptService",
input=UpdateScriptRequest,
output=UpdateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.update_script,
),
"/soulfire.v1.ScriptService/DeleteScript": EndpointSync.unary(
method=MethodInfo(
name="DeleteScript",
service_name="soulfire.v1.ScriptService",
input=DeleteScriptRequest,
output=DeleteScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.delete_script,
),
"/soulfire.v1.ScriptService/ListScripts": EndpointSync.unary(
method=MethodInfo(
name="ListScripts",
service_name="soulfire.v1.ScriptService",
input=ListScriptsRequest,
output=ListScriptsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.list_scripts,
),
"/soulfire.v1.ScriptService/ActivateScript": EndpointSync.server_stream(
method=MethodInfo(
name="ActivateScript",
service_name="soulfire.v1.ScriptService",
input=ActivateScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.activate_script,
),
"/soulfire.v1.ScriptService/DeactivateScript": EndpointSync.unary(
method=MethodInfo(
name="DeactivateScript",
service_name="soulfire.v1.ScriptService",
input=DeactivateScriptRequest,
output=DeactivateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.deactivate_script,
),
"/soulfire.v1.ScriptService/GetScriptStatus": EndpointSync.unary(
method=MethodInfo(
name="GetScriptStatus",
service_name="soulfire.v1.ScriptService",
input=GetScriptStatusRequest,
output=GetScriptStatusResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_script_status,
),
"/soulfire.v1.ScriptService/SubscribeScriptLogs": EndpointSync.server_stream(
method=MethodInfo(
name="SubscribeScriptLogs",
service_name="soulfire.v1.ScriptService",
input=SubscribeScriptLogsRequest,
output=ScriptLogEntry,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.subscribe_script_logs,
),
"/soulfire.v1.ScriptService/GetNodeTypes": EndpointSync.unary(
method=MethodInfo(
name="GetNodeTypes",
service_name="soulfire.v1.ScriptService",
input=GetNodeTypesRequest,
output=GetNodeTypesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_node_types,
),
"/soulfire.v1.ScriptService/GetRegistryData": EndpointSync.unary(
method=MethodInfo(
name="GetRegistryData",
service_name="soulfire.v1.ScriptService",
input=GetRegistryDataRequest,
output=GetRegistryDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_registry_data,
),
"/soulfire.v1.ScriptService/ValidateScript": EndpointSync.unary(
method=MethodInfo(
name="ValidateScript",
service_name="soulfire.v1.ScriptService",
input=ValidateScriptRequest,
output=ValidateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.validate_script,
),
"/soulfire.v1.ScriptService/DryRunScript": EndpointSync.server_stream(
method=MethodInfo(
name="DryRunScript",
service_name="soulfire.v1.ScriptService",
input=DryRunScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.dry_run_script,
),
},
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.ScriptService"
[docs]
class ScriptServiceClientSync(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 create_script(
self,
request: CreateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CreateScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CreateScript",
service_name="soulfire.v1.ScriptService",
input=CreateScriptRequest,
output=CreateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_script(
self,
request: GetScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetScript",
service_name="soulfire.v1.ScriptService",
input=GetScriptRequest,
output=GetScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def update_script(
self,
request: UpdateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateScript",
service_name="soulfire.v1.ScriptService",
input=UpdateScriptRequest,
output=UpdateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def delete_script(
self,
request: DeleteScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DeleteScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="DeleteScript",
service_name="soulfire.v1.ScriptService",
input=DeleteScriptRequest,
output=DeleteScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def list_scripts(
self,
request: ListScriptsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListScriptsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ListScripts",
service_name="soulfire.v1.ScriptService",
input=ListScriptsRequest,
output=ListScriptsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def activate_script(
self,
request: ActivateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[ScriptEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="ActivateScript",
service_name="soulfire.v1.ScriptService",
input=ActivateScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def deactivate_script(
self,
request: DeactivateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DeactivateScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="DeactivateScript",
service_name="soulfire.v1.ScriptService",
input=DeactivateScriptRequest,
output=DeactivateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_script_status(
self,
request: GetScriptStatusRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetScriptStatusResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetScriptStatus",
service_name="soulfire.v1.ScriptService",
input=GetScriptStatusRequest,
output=GetScriptStatusResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def subscribe_script_logs(
self,
request: SubscribeScriptLogsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[ScriptLogEntry]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="SubscribeScriptLogs",
service_name="soulfire.v1.ScriptService",
input=SubscribeScriptLogsRequest,
output=ScriptLogEntry,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_node_types(
self,
request: GetNodeTypesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetNodeTypesResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetNodeTypes",
service_name="soulfire.v1.ScriptService",
input=GetNodeTypesRequest,
output=GetNodeTypesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_registry_data(
self,
request: GetRegistryDataRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetRegistryDataResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetRegistryData",
service_name="soulfire.v1.ScriptService",
input=GetRegistryDataRequest,
output=GetRegistryDataResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def validate_script(
self,
request: ValidateScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ValidateScriptResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ValidateScript",
service_name="soulfire.v1.ScriptService",
input=ValidateScriptRequest,
output=ValidateScriptResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def dry_run_script(
self,
request: DryRunScriptRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[ScriptEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="DryRunScript",
service_name="soulfire.v1.ScriptService",
input=DryRunScriptRequest,
output=ScriptEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)