# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: ibc/applications/transfer/v1/authz.proto, ibc/applications/transfer/v1/genesis.proto, ibc/applications/transfer/v1/query.proto, ibc/applications/transfer/v1/transfer.proto, ibc/applications/transfer/v1/tx.proto
# plugin: python-betterproto
# This file has been @generated
from dataclasses import dataclass
from typing import (
TYPE_CHECKING,
Dict,
List,
Optional,
)
import betterproto
import grpclib
from betterproto.grpc.grpclib_server import ServiceBase
from .....cosmos.base import v1beta1 as ____cosmos_base_v1_beta1__
from .....cosmos.base.query import v1beta1 as ____cosmos_base_query_v1_beta1__
from ....core.client import v1 as ___core_client_v1__
if TYPE_CHECKING:
import grpclib.server
from betterproto.grpc.grpclib_client import MetadataLike
from grpclib.metadata import Deadline
[docs]
@dataclass(eq=False, repr=False)
class Allocation(betterproto.Message):
"""Allocation defines the spend limit for a particular port and channel"""
source_port: str = betterproto.string_field(1)
"""the port on which the packet will be sent"""
source_channel: str = betterproto.string_field(2)
"""the channel by which the packet will be sent"""
spend_limit: List["____cosmos_base_v1_beta1__.Coin"] = betterproto.message_field(3)
"""spend limitation on the channel"""
allow_list: List[str] = betterproto.string_field(4)
"""allow list of receivers, an empty allow list permits any receiver address"""
allowed_packet_data: List[str] = betterproto.string_field(5)
"""
allow list of memo strings, an empty list prohibits all memo strings;
a list only with "*" permits any memo string
"""
[docs]
@dataclass(eq=False, repr=False)
class TransferAuthorization(betterproto.Message):
"""
TransferAuthorization allows the grantee to spend up to spend_limit coins from
the granter's account for ibc transfer on a specific channel
"""
allocations: List["Allocation"] = betterproto.message_field(1)
"""port and channel amounts"""
[docs]
@dataclass(eq=False, repr=False)
class DenomTrace(betterproto.Message):
"""
DenomTrace contains the base denomination for ICS20 fungible tokens and the
source tracing information path.
"""
path: str = betterproto.string_field(1)
"""
path defines the chain of port/channel identifiers used for tracing the
source of the fungible token.
"""
base_denom: str = betterproto.string_field(2)
"""base denomination of the relayed fungible token."""
[docs]
@dataclass(eq=False, repr=False)
class Params(betterproto.Message):
"""
Params defines the set of IBC transfer parameters.
NOTE: To prevent a single token from being transferred, set the
TransfersEnabled parameter to true and then set the bank module's SendEnabled
parameter for the denomination to false.
"""
send_enabled: bool = betterproto.bool_field(1)
"""
send_enabled enables or disables all cross-chain token transfers from this
chain.
"""
receive_enabled: bool = betterproto.bool_field(2)
"""
receive_enabled enables or disables all cross-chain token transfers to this
chain.
"""
[docs]
@dataclass(eq=False, repr=False)
class GenesisState(betterproto.Message):
"""GenesisState defines the ibc-transfer genesis state"""
port_id: str = betterproto.string_field(1)
denom_traces: List["DenomTrace"] = betterproto.message_field(2)
params: "Params" = betterproto.message_field(3)
total_escrowed: List["____cosmos_base_v1_beta1__.Coin"] = betterproto.message_field(
4
)
"""
total_escrowed contains the total amount of tokens escrowed
by the transfer module
"""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomTraceRequest(betterproto.Message):
"""
QueryDenomTraceRequest is the request type for the Query/DenomTrace RPC
method
"""
hash: str = betterproto.string_field(1)
"""
hash (in hex format) or denom (full denom with ibc prefix) of the denomination trace
information.
"""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomTraceResponse(betterproto.Message):
"""
QueryDenomTraceResponse is the response type for the Query/DenomTrace RPC
method.
"""
denom_trace: "DenomTrace" = betterproto.message_field(1)
"""denom_trace returns the requested denomination trace information."""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomTracesRequest(betterproto.Message):
"""
QueryConnectionsRequest is the request type for the Query/DenomTraces RPC
method
"""
pagination: "____cosmos_base_query_v1_beta1__.PageRequest" = (
betterproto.message_field(1)
)
"""pagination defines an optional pagination for the request."""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomTracesResponse(betterproto.Message):
"""
QueryConnectionsResponse is the response type for the Query/DenomTraces RPC
method.
"""
denom_traces: List["DenomTrace"] = betterproto.message_field(1)
"""denom_traces returns all denominations trace information."""
pagination: "____cosmos_base_query_v1_beta1__.PageResponse" = (
betterproto.message_field(2)
)
"""pagination defines the pagination in the response."""
[docs]
@dataclass(eq=False, repr=False)
class QueryParamsRequest(betterproto.Message):
"""QueryParamsRequest is the request type for the Query/Params RPC method."""
pass
[docs]
@dataclass(eq=False, repr=False)
class QueryParamsResponse(betterproto.Message):
"""QueryParamsResponse is the response type for the Query/Params RPC method."""
params: "Params" = betterproto.message_field(1)
"""params defines the parameters of the module."""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomHashRequest(betterproto.Message):
"""
QueryDenomHashRequest is the request type for the Query/DenomHash RPC
method
"""
trace: str = betterproto.string_field(1)
"""The denomination trace ([port_id]/[channel_id])+/[denom]"""
[docs]
@dataclass(eq=False, repr=False)
class QueryDenomHashResponse(betterproto.Message):
"""
QueryDenomHashResponse is the response type for the Query/DenomHash RPC
method.
"""
hash: str = betterproto.string_field(1)
"""hash (in hex format) of the denomination trace information."""
[docs]
@dataclass(eq=False, repr=False)
class QueryEscrowAddressRequest(betterproto.Message):
"""
QueryEscrowAddressRequest is the request type for the EscrowAddress RPC method.
"""
port_id: str = betterproto.string_field(1)
"""unique port identifier"""
channel_id: str = betterproto.string_field(2)
"""unique channel identifier"""
[docs]
@dataclass(eq=False, repr=False)
class QueryEscrowAddressResponse(betterproto.Message):
"""
QueryEscrowAddressResponse is the response type of the EscrowAddress RPC method.
"""
escrow_address: str = betterproto.string_field(1)
"""the escrow account address"""
[docs]
@dataclass(eq=False, repr=False)
class QueryTotalEscrowForDenomRequest(betterproto.Message):
"""
QueryTotalEscrowForDenomRequest is the request type for TotalEscrowForDenom RPC
method.
"""
denom: str = betterproto.string_field(1)
[docs]
@dataclass(eq=False, repr=False)
class QueryTotalEscrowForDenomResponse(betterproto.Message):
"""
QueryTotalEscrowForDenomResponse is the response type for TotalEscrowForDenom RPC
method.
"""
amount: "____cosmos_base_v1_beta1__.Coin" = betterproto.message_field(1)
[docs]
@dataclass(eq=False, repr=False)
class MsgTransfer(betterproto.Message):
"""
MsgTransfer defines a msg to transfer fungible tokens (i.e Coins) between
ICS20 enabled chains. See ICS Spec here:
https://github.com/cosmos/ibc/tree/main/spec/app/ics-020-fungible-token-transfer
"""
source_port: str = betterproto.string_field(1)
"""the port on which the packet will be sent"""
source_channel: str = betterproto.string_field(2)
"""the channel by which the packet will be sent"""
token: "____cosmos_base_v1_beta1__.Coin" = betterproto.message_field(3)
"""the tokens to be transferred"""
sender: str = betterproto.string_field(4)
"""the sender address"""
receiver: str = betterproto.string_field(5)
"""the recipient address on the destination chain"""
timeout_height: "___core_client_v1__.Height" = betterproto.message_field(6)
"""
Timeout height relative to the current block height.
The timeout is disabled when set to 0.
"""
timeout_timestamp: int = betterproto.uint64_field(7)
"""
Timeout timestamp in absolute nanoseconds since unix epoch.
The timeout is disabled when set to 0.
"""
memo: str = betterproto.string_field(8)
"""optional memo"""
[docs]
@dataclass(eq=False, repr=False)
class MsgTransferResponse(betterproto.Message):
"""MsgTransferResponse defines the Msg/Transfer response type."""
sequence: int = betterproto.uint64_field(1)
"""sequence number of the transfer packet sent"""
[docs]
@dataclass(eq=False, repr=False)
class MsgUpdateParams(betterproto.Message):
"""MsgUpdateParams is the Msg/UpdateParams request type."""
signer: str = betterproto.string_field(1)
"""signer address"""
params: "Params" = betterproto.message_field(2)
"""
params defines the transfer parameters to update.
NOTE: All parameters must be supplied.
"""
[docs]
@dataclass(eq=False, repr=False)
class MsgUpdateParamsResponse(betterproto.Message):
"""
MsgUpdateParamsResponse defines the response structure for executing a
MsgUpdateParams message.
"""
pass
[docs]
class QueryStub(betterproto.ServiceStub):
[docs]
async def denom_traces(
self,
query_denom_traces_request: "QueryDenomTracesRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryDenomTracesResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/DenomTraces",
query_denom_traces_request,
QueryDenomTracesResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def denom_trace(
self,
query_denom_trace_request: "QueryDenomTraceRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryDenomTraceResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/DenomTrace",
query_denom_trace_request,
QueryDenomTraceResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def params(
self,
query_params_request: "QueryParamsRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryParamsResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/Params",
query_params_request,
QueryParamsResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def denom_hash(
self,
query_denom_hash_request: "QueryDenomHashRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryDenomHashResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/DenomHash",
query_denom_hash_request,
QueryDenomHashResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def escrow_address(
self,
query_escrow_address_request: "QueryEscrowAddressRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryEscrowAddressResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/EscrowAddress",
query_escrow_address_request,
QueryEscrowAddressResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def total_escrow_for_denom(
self,
query_total_escrow_for_denom_request: "QueryTotalEscrowForDenomRequest",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "QueryTotalEscrowForDenomResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Query/TotalEscrowForDenom",
query_total_escrow_for_denom_request,
QueryTotalEscrowForDenomResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
class MsgStub(betterproto.ServiceStub):
[docs]
async def transfer(
self,
msg_transfer: "MsgTransfer",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "MsgTransferResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Msg/Transfer",
msg_transfer,
MsgTransferResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
async def update_params(
self,
msg_update_params: "MsgUpdateParams",
*,
timeout: Optional[float] = None,
deadline: Optional["Deadline"] = None,
metadata: Optional["MetadataLike"] = None,
) -> "MsgUpdateParamsResponse":
return await self._unary_unary(
"/ibc.applications.transfer.v1.Msg/UpdateParams",
msg_update_params,
MsgUpdateParamsResponse,
timeout=timeout,
deadline=deadline,
metadata=metadata,
)
[docs]
class QueryBase(ServiceBase):
[docs]
async def denom_traces(
self, query_denom_traces_request: "QueryDenomTracesRequest"
) -> "QueryDenomTracesResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def denom_trace(
self, query_denom_trace_request: "QueryDenomTraceRequest"
) -> "QueryDenomTraceResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def params(
self, query_params_request: "QueryParamsRequest"
) -> "QueryParamsResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def denom_hash(
self, query_denom_hash_request: "QueryDenomHashRequest"
) -> "QueryDenomHashResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def escrow_address(
self, query_escrow_address_request: "QueryEscrowAddressRequest"
) -> "QueryEscrowAddressResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def total_escrow_for_denom(
self, query_total_escrow_for_denom_request: "QueryTotalEscrowForDenomRequest"
) -> "QueryTotalEscrowForDenomResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_denom_traces(
self,
stream: "grpclib.server.Stream[QueryDenomTracesRequest, QueryDenomTracesResponse]",
) -> None:
request = await stream.recv_message()
response = await self.denom_traces(request)
await stream.send_message(response)
async def __rpc_denom_trace(
self,
stream: "grpclib.server.Stream[QueryDenomTraceRequest, QueryDenomTraceResponse]",
) -> None:
request = await stream.recv_message()
response = await self.denom_trace(request)
await stream.send_message(response)
async def __rpc_params(
self, stream: "grpclib.server.Stream[QueryParamsRequest, QueryParamsResponse]"
) -> None:
request = await stream.recv_message()
response = await self.params(request)
await stream.send_message(response)
async def __rpc_denom_hash(
self,
stream: "grpclib.server.Stream[QueryDenomHashRequest, QueryDenomHashResponse]",
) -> None:
request = await stream.recv_message()
response = await self.denom_hash(request)
await stream.send_message(response)
async def __rpc_escrow_address(
self,
stream: "grpclib.server.Stream[QueryEscrowAddressRequest, QueryEscrowAddressResponse]",
) -> None:
request = await stream.recv_message()
response = await self.escrow_address(request)
await stream.send_message(response)
async def __rpc_total_escrow_for_denom(
self,
stream: "grpclib.server.Stream[QueryTotalEscrowForDenomRequest, QueryTotalEscrowForDenomResponse]",
) -> None:
request = await stream.recv_message()
response = await self.total_escrow_for_denom(request)
await stream.send_message(response)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/ibc.applications.transfer.v1.Query/DenomTraces": grpclib.const.Handler(
self.__rpc_denom_traces,
grpclib.const.Cardinality.UNARY_UNARY,
QueryDenomTracesRequest,
QueryDenomTracesResponse,
),
"/ibc.applications.transfer.v1.Query/DenomTrace": grpclib.const.Handler(
self.__rpc_denom_trace,
grpclib.const.Cardinality.UNARY_UNARY,
QueryDenomTraceRequest,
QueryDenomTraceResponse,
),
"/ibc.applications.transfer.v1.Query/Params": grpclib.const.Handler(
self.__rpc_params,
grpclib.const.Cardinality.UNARY_UNARY,
QueryParamsRequest,
QueryParamsResponse,
),
"/ibc.applications.transfer.v1.Query/DenomHash": grpclib.const.Handler(
self.__rpc_denom_hash,
grpclib.const.Cardinality.UNARY_UNARY,
QueryDenomHashRequest,
QueryDenomHashResponse,
),
"/ibc.applications.transfer.v1.Query/EscrowAddress": grpclib.const.Handler(
self.__rpc_escrow_address,
grpclib.const.Cardinality.UNARY_UNARY,
QueryEscrowAddressRequest,
QueryEscrowAddressResponse,
),
"/ibc.applications.transfer.v1.Query/TotalEscrowForDenom": grpclib.const.Handler(
self.__rpc_total_escrow_for_denom,
grpclib.const.Cardinality.UNARY_UNARY,
QueryTotalEscrowForDenomRequest,
QueryTotalEscrowForDenomResponse,
),
}
[docs]
class MsgBase(ServiceBase):
[docs]
async def transfer(self, msg_transfer: "MsgTransfer") -> "MsgTransferResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
[docs]
async def update_params(
self, msg_update_params: "MsgUpdateParams"
) -> "MsgUpdateParamsResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_transfer(
self, stream: "grpclib.server.Stream[MsgTransfer, MsgTransferResponse]"
) -> None:
request = await stream.recv_message()
response = await self.transfer(request)
await stream.send_message(response)
async def __rpc_update_params(
self, stream: "grpclib.server.Stream[MsgUpdateParams, MsgUpdateParamsResponse]"
) -> None:
request = await stream.recv_message()
response = await self.update_params(request)
await stream.send_message(response)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/ibc.applications.transfer.v1.Msg/Transfer": grpclib.const.Handler(
self.__rpc_transfer,
grpclib.const.Cardinality.UNARY_UNARY,
MsgTransfer,
MsgTransferResponse,
),
"/ibc.applications.transfer.v1.Msg/UpdateParams": grpclib.const.Handler(
self.__rpc_update_params,
grpclib.const.Cardinality.UNARY_UNARY,
MsgUpdateParams,
MsgUpdateParamsResponse,
),
}