# Generated from soulfire/inventory.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 .inventory_pb2 import CloseSemanticContainerRequest, CountItemsRequest, CountItemsResponse, EquipItemRequest, FindInventorySlotsRequest, FindInventorySlotsResponse, GetContainerSnapshotRequest, GetContainerSnapshotResponse, InventoryMutationResponse, MoveInventoryItemRequest, OpenBlockContainerRequest, RankInventoryItemsRequest, RankInventoryItemsResponse, SelectHotbarItemRequest, TossItemsRequest, TransferItemsRequest, UnequipItemRequest
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 InventoryService(Protocol):
[docs]
async def get_container_snapshot(self, request: GetContainerSnapshotRequest, ctx: RequestContext[GetContainerSnapshotRequest, GetContainerSnapshotResponse]) -> GetContainerSnapshotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def count_items(self, request: CountItemsRequest, ctx: RequestContext[CountItemsRequest, CountItemsResponse]) -> CountItemsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def find_inventory_slots(self, request: FindInventorySlotsRequest, ctx: RequestContext[FindInventorySlotsRequest, FindInventorySlotsResponse]) -> FindInventorySlotsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def rank_inventory_items(self, request: RankInventoryItemsRequest, ctx: RequestContext[RankInventoryItemsRequest, RankInventoryItemsResponse]) -> RankInventoryItemsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def move_inventory_item(self, request: MoveInventoryItemRequest, ctx: RequestContext[MoveInventoryItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def transfer_items(self, request: TransferItemsRequest, ctx: RequestContext[TransferItemsRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def toss_items(self, request: TossItemsRequest, ctx: RequestContext[TossItemsRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def select_hotbar_item(self, request: SelectHotbarItemRequest, ctx: RequestContext[SelectHotbarItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def equip_item(self, request: EquipItemRequest, ctx: RequestContext[EquipItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def unequip_item(self, request: UnequipItemRequest, ctx: RequestContext[UnequipItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def open_block_container(self, request: OpenBlockContainerRequest, ctx: RequestContext[OpenBlockContainerRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def close_semantic_container(self, request: CloseSemanticContainerRequest, ctx: RequestContext[CloseSemanticContainerRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class InventoryServiceASGIApplication(ConnectASGIApplication[InventoryService]):
def __init__(
self,
service: InventoryService | AsyncGenerator[InventoryService],
*,
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.InventoryService/GetContainerSnapshot": Endpoint.unary(
method=MethodInfo(
name="GetContainerSnapshot",
service_name="soulfire.v1.InventoryService",
input=GetContainerSnapshotRequest,
output=GetContainerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_container_snapshot,
),
"/soulfire.v1.InventoryService/CountItems": Endpoint.unary(
method=MethodInfo(
name="CountItems",
service_name="soulfire.v1.InventoryService",
input=CountItemsRequest,
output=CountItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.count_items,
),
"/soulfire.v1.InventoryService/FindInventorySlots": Endpoint.unary(
method=MethodInfo(
name="FindInventorySlots",
service_name="soulfire.v1.InventoryService",
input=FindInventorySlotsRequest,
output=FindInventorySlotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.find_inventory_slots,
),
"/soulfire.v1.InventoryService/RankInventoryItems": Endpoint.unary(
method=MethodInfo(
name="RankInventoryItems",
service_name="soulfire.v1.InventoryService",
input=RankInventoryItemsRequest,
output=RankInventoryItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.rank_inventory_items,
),
"/soulfire.v1.InventoryService/MoveInventoryItem": Endpoint.unary(
method=MethodInfo(
name="MoveInventoryItem",
service_name="soulfire.v1.InventoryService",
input=MoveInventoryItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.move_inventory_item,
),
"/soulfire.v1.InventoryService/TransferItems": Endpoint.unary(
method=MethodInfo(
name="TransferItems",
service_name="soulfire.v1.InventoryService",
input=TransferItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.transfer_items,
),
"/soulfire.v1.InventoryService/TossItems": Endpoint.unary(
method=MethodInfo(
name="TossItems",
service_name="soulfire.v1.InventoryService",
input=TossItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.toss_items,
),
"/soulfire.v1.InventoryService/SelectHotbarItem": Endpoint.unary(
method=MethodInfo(
name="SelectHotbarItem",
service_name="soulfire.v1.InventoryService",
input=SelectHotbarItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.select_hotbar_item,
),
"/soulfire.v1.InventoryService/EquipItem": Endpoint.unary(
method=MethodInfo(
name="EquipItem",
service_name="soulfire.v1.InventoryService",
input=EquipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.equip_item,
),
"/soulfire.v1.InventoryService/UnequipItem": Endpoint.unary(
method=MethodInfo(
name="UnequipItem",
service_name="soulfire.v1.InventoryService",
input=UnequipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.unequip_item,
),
"/soulfire.v1.InventoryService/OpenBlockContainer": Endpoint.unary(
method=MethodInfo(
name="OpenBlockContainer",
service_name="soulfire.v1.InventoryService",
input=OpenBlockContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.open_block_container,
),
"/soulfire.v1.InventoryService/CloseSemanticContainer": Endpoint.unary(
method=MethodInfo(
name="CloseSemanticContainer",
service_name="soulfire.v1.InventoryService",
input=CloseSemanticContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.close_semantic_container,
),
},
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.InventoryService"
[docs]
class InventoryServiceClient(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_container_snapshot(
self,
request: GetContainerSnapshotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetContainerSnapshotResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetContainerSnapshot",
service_name="soulfire.v1.InventoryService",
input=GetContainerSnapshotRequest,
output=GetContainerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def count_items(
self,
request: CountItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CountItemsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CountItems",
service_name="soulfire.v1.InventoryService",
input=CountItemsRequest,
output=CountItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def find_inventory_slots(
self,
request: FindInventorySlotsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> FindInventorySlotsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="FindInventorySlots",
service_name="soulfire.v1.InventoryService",
input=FindInventorySlotsRequest,
output=FindInventorySlotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def rank_inventory_items(
self,
request: RankInventoryItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RankInventoryItemsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="RankInventoryItems",
service_name="soulfire.v1.InventoryService",
input=RankInventoryItemsRequest,
output=RankInventoryItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def move_inventory_item(
self,
request: MoveInventoryItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="MoveInventoryItem",
service_name="soulfire.v1.InventoryService",
input=MoveInventoryItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def transfer_items(
self,
request: TransferItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="TransferItems",
service_name="soulfire.v1.InventoryService",
input=TransferItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def toss_items(
self,
request: TossItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="TossItems",
service_name="soulfire.v1.InventoryService",
input=TossItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def select_hotbar_item(
self,
request: SelectHotbarItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SelectHotbarItem",
service_name="soulfire.v1.InventoryService",
input=SelectHotbarItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def equip_item(
self,
request: EquipItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="EquipItem",
service_name="soulfire.v1.InventoryService",
input=EquipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def unequip_item(
self,
request: UnequipItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UnequipItem",
service_name="soulfire.v1.InventoryService",
input=UnequipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def open_block_container(
self,
request: OpenBlockContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="OpenBlockContainer",
service_name="soulfire.v1.InventoryService",
input=OpenBlockContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def close_semantic_container(
self,
request: CloseSemanticContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CloseSemanticContainer",
service_name="soulfire.v1.InventoryService",
input=CloseSemanticContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class InventoryServiceSync(Protocol):
[docs]
def get_container_snapshot(self, request: GetContainerSnapshotRequest, ctx: RequestContext[GetContainerSnapshotRequest, GetContainerSnapshotResponse]) -> GetContainerSnapshotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def count_items(self, request: CountItemsRequest, ctx: RequestContext[CountItemsRequest, CountItemsResponse]) -> CountItemsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def find_inventory_slots(self, request: FindInventorySlotsRequest, ctx: RequestContext[FindInventorySlotsRequest, FindInventorySlotsResponse]) -> FindInventorySlotsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def rank_inventory_items(self, request: RankInventoryItemsRequest, ctx: RequestContext[RankInventoryItemsRequest, RankInventoryItemsResponse]) -> RankInventoryItemsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def move_inventory_item(self, request: MoveInventoryItemRequest, ctx: RequestContext[MoveInventoryItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def transfer_items(self, request: TransferItemsRequest, ctx: RequestContext[TransferItemsRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def toss_items(self, request: TossItemsRequest, ctx: RequestContext[TossItemsRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def select_hotbar_item(self, request: SelectHotbarItemRequest, ctx: RequestContext[SelectHotbarItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def equip_item(self, request: EquipItemRequest, ctx: RequestContext[EquipItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def unequip_item(self, request: UnequipItemRequest, ctx: RequestContext[UnequipItemRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def open_block_container(self, request: OpenBlockContainerRequest, ctx: RequestContext[OpenBlockContainerRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def close_semantic_container(self, request: CloseSemanticContainerRequest, ctx: RequestContext[CloseSemanticContainerRequest, InventoryMutationResponse]) -> InventoryMutationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class InventoryServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: InventoryServiceSync,
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.InventoryService/GetContainerSnapshot": EndpointSync.unary(
method=MethodInfo(
name="GetContainerSnapshot",
service_name="soulfire.v1.InventoryService",
input=GetContainerSnapshotRequest,
output=GetContainerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_container_snapshot,
),
"/soulfire.v1.InventoryService/CountItems": EndpointSync.unary(
method=MethodInfo(
name="CountItems",
service_name="soulfire.v1.InventoryService",
input=CountItemsRequest,
output=CountItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.count_items,
),
"/soulfire.v1.InventoryService/FindInventorySlots": EndpointSync.unary(
method=MethodInfo(
name="FindInventorySlots",
service_name="soulfire.v1.InventoryService",
input=FindInventorySlotsRequest,
output=FindInventorySlotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.find_inventory_slots,
),
"/soulfire.v1.InventoryService/RankInventoryItems": EndpointSync.unary(
method=MethodInfo(
name="RankInventoryItems",
service_name="soulfire.v1.InventoryService",
input=RankInventoryItemsRequest,
output=RankInventoryItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.rank_inventory_items,
),
"/soulfire.v1.InventoryService/MoveInventoryItem": EndpointSync.unary(
method=MethodInfo(
name="MoveInventoryItem",
service_name="soulfire.v1.InventoryService",
input=MoveInventoryItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.move_inventory_item,
),
"/soulfire.v1.InventoryService/TransferItems": EndpointSync.unary(
method=MethodInfo(
name="TransferItems",
service_name="soulfire.v1.InventoryService",
input=TransferItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.transfer_items,
),
"/soulfire.v1.InventoryService/TossItems": EndpointSync.unary(
method=MethodInfo(
name="TossItems",
service_name="soulfire.v1.InventoryService",
input=TossItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.toss_items,
),
"/soulfire.v1.InventoryService/SelectHotbarItem": EndpointSync.unary(
method=MethodInfo(
name="SelectHotbarItem",
service_name="soulfire.v1.InventoryService",
input=SelectHotbarItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.select_hotbar_item,
),
"/soulfire.v1.InventoryService/EquipItem": EndpointSync.unary(
method=MethodInfo(
name="EquipItem",
service_name="soulfire.v1.InventoryService",
input=EquipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.equip_item,
),
"/soulfire.v1.InventoryService/UnequipItem": EndpointSync.unary(
method=MethodInfo(
name="UnequipItem",
service_name="soulfire.v1.InventoryService",
input=UnequipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.unequip_item,
),
"/soulfire.v1.InventoryService/OpenBlockContainer": EndpointSync.unary(
method=MethodInfo(
name="OpenBlockContainer",
service_name="soulfire.v1.InventoryService",
input=OpenBlockContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.open_block_container,
),
"/soulfire.v1.InventoryService/CloseSemanticContainer": EndpointSync.unary(
method=MethodInfo(
name="CloseSemanticContainer",
service_name="soulfire.v1.InventoryService",
input=CloseSemanticContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.close_semantic_container,
),
},
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.InventoryService"
[docs]
class InventoryServiceClientSync(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_container_snapshot(
self,
request: GetContainerSnapshotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetContainerSnapshotResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetContainerSnapshot",
service_name="soulfire.v1.InventoryService",
input=GetContainerSnapshotRequest,
output=GetContainerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def count_items(
self,
request: CountItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CountItemsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CountItems",
service_name="soulfire.v1.InventoryService",
input=CountItemsRequest,
output=CountItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def find_inventory_slots(
self,
request: FindInventorySlotsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> FindInventorySlotsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="FindInventorySlots",
service_name="soulfire.v1.InventoryService",
input=FindInventorySlotsRequest,
output=FindInventorySlotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def rank_inventory_items(
self,
request: RankInventoryItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RankInventoryItemsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="RankInventoryItems",
service_name="soulfire.v1.InventoryService",
input=RankInventoryItemsRequest,
output=RankInventoryItemsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def move_inventory_item(
self,
request: MoveInventoryItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="MoveInventoryItem",
service_name="soulfire.v1.InventoryService",
input=MoveInventoryItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def transfer_items(
self,
request: TransferItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="TransferItems",
service_name="soulfire.v1.InventoryService",
input=TransferItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def toss_items(
self,
request: TossItemsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="TossItems",
service_name="soulfire.v1.InventoryService",
input=TossItemsRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def select_hotbar_item(
self,
request: SelectHotbarItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SelectHotbarItem",
service_name="soulfire.v1.InventoryService",
input=SelectHotbarItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def equip_item(
self,
request: EquipItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="EquipItem",
service_name="soulfire.v1.InventoryService",
input=EquipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def unequip_item(
self,
request: UnequipItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UnequipItem",
service_name="soulfire.v1.InventoryService",
input=UnequipItemRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def open_block_container(
self,
request: OpenBlockContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="OpenBlockContainer",
service_name="soulfire.v1.InventoryService",
input=OpenBlockContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def close_semantic_container(
self,
request: CloseSemanticContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InventoryMutationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CloseSemanticContainer",
service_name="soulfire.v1.InventoryService",
input=CloseSemanticContainerRequest,
output=InventoryMutationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)