# Generated by the protocol buffer compiler. DO NOT EDIT!
# sources: cosmos/gov/v1beta1/genesis.proto, cosmos/gov/v1beta1/gov.proto, cosmos/gov/v1beta1/query.proto, cosmos/gov/v1beta1/tx.proto
# plugin: python-betterproto
import warnings
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import betterproto
from betterproto.grpc.grpclib_server import ServiceBase
import grpclib
[docs]class VoteOption(betterproto.Enum):
"""
VoteOption enumerates the valid vote options for a given governance
proposal.
"""
# VOTE_OPTION_UNSPECIFIED defines a no-op vote option.
VOTE_OPTION_UNSPECIFIED = 0
# VOTE_OPTION_YES defines a yes vote option.
VOTE_OPTION_YES = 1
# VOTE_OPTION_ABSTAIN defines an abstain vote option.
VOTE_OPTION_ABSTAIN = 2
# VOTE_OPTION_NO defines a no vote option.
VOTE_OPTION_NO = 3
# VOTE_OPTION_NO_WITH_VETO defines a no with veto vote option.
VOTE_OPTION_NO_WITH_VETO = 4
class ProposalStatus(betterproto.Enum):
"""ProposalStatus enumerates the valid statuses of a proposal."""
# PROPOSAL_STATUS_UNSPECIFIED defines the default propopsal status.
PROPOSAL_STATUS_UNSPECIFIED = 0
# PROPOSAL_STATUS_DEPOSIT_PERIOD defines a proposal status during the deposit
# period.
PROPOSAL_STATUS_DEPOSIT_PERIOD = 1
# PROPOSAL_STATUS_VOTING_PERIOD defines a proposal status during the voting
# period.
PROPOSAL_STATUS_VOTING_PERIOD = 2
# PROPOSAL_STATUS_PASSED defines a proposal status of a proposal that has
# passed.
PROPOSAL_STATUS_PASSED = 3
# PROPOSAL_STATUS_REJECTED defines a proposal status of a proposal that has
# been rejected.
PROPOSAL_STATUS_REJECTED = 4
# PROPOSAL_STATUS_FAILED defines a proposal status of a proposal that has
# failed.
PROPOSAL_STATUS_FAILED = 5
@dataclass(eq=False, repr=False)
class WeightedVoteOption(betterproto.Message):
"""WeightedVoteOption defines a unit of vote for vote split."""
option: "VoteOption" = betterproto.enum_field(1)
weight: str = betterproto.string_field(2)
@dataclass(eq=False, repr=False)
class TextProposal(betterproto.Message):
"""
TextProposal defines a standard text proposal whose changes need to be
manually updated in case of approval.
"""
title: str = betterproto.string_field(1)
description: str = betterproto.string_field(2)
@dataclass(eq=False, repr=False)
class Deposit(betterproto.Message):
"""
Deposit defines an amount deposited by an account address to an active
proposal.
"""
proposal_id: int = betterproto.uint64_field(1)
depositor: str = betterproto.string_field(2)
amount: List["__base_v1_beta1__.Coin"] = betterproto.message_field(3)
@dataclass(eq=False, repr=False)
class Proposal(betterproto.Message):
"""Proposal defines the core field members of a governance proposal."""
proposal_id: int = betterproto.uint64_field(1)
content: "betterproto_lib_google_protobuf.Any" = betterproto.message_field(2)
status: "ProposalStatus" = betterproto.enum_field(3)
final_tally_result: "TallyResult" = betterproto.message_field(4)
submit_time: datetime = betterproto.message_field(5)
deposit_end_time: datetime = betterproto.message_field(6)
total_deposit: List["__base_v1_beta1__.Coin"] = betterproto.message_field(7)
voting_start_time: datetime = betterproto.message_field(8)
voting_end_time: datetime = betterproto.message_field(9)
@dataclass(eq=False, repr=False)
class TallyResult(betterproto.Message):
"""TallyResult defines a standard tally for a governance proposal."""
yes: str = betterproto.string_field(1)
abstain: str = betterproto.string_field(2)
no: str = betterproto.string_field(3)
no_with_veto: str = betterproto.string_field(4)
@dataclass(eq=False, repr=False)
class Vote(betterproto.Message):
"""
Vote defines a vote on a governance proposal. A Vote consists of a proposal
ID, the voter, and the vote option.
"""
proposal_id: int = betterproto.uint64_field(1)
voter: str = betterproto.string_field(2)
# Deprecated: Prefer to use `options` instead. This field is set in queries
# if and only if `len(options) == 1` and that option has weight 1. In all
# other cases, this field will default to VOTE_OPTION_UNSPECIFIED.
option: "VoteOption" = betterproto.enum_field(3)
options: List["WeightedVoteOption"] = betterproto.message_field(4)
def __post_init__(self) -> None:
super().__post_init__()
if self.option:
warnings.warn("Vote.option is deprecated", DeprecationWarning)
@dataclass(eq=False, repr=False)
class DepositParams(betterproto.Message):
"""
DepositParams defines the params for deposits on governance proposals.
"""
# Minimum deposit for a proposal to enter voting period.
min_deposit: List["__base_v1_beta1__.Coin"] = betterproto.message_field(1)
# Maximum period for Atom holders to deposit on a proposal. Initial value: 2
# months.
max_deposit_period: timedelta = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class VotingParams(betterproto.Message):
"""VotingParams defines the params for voting on governance proposals."""
# Length of the voting period.
voting_period: timedelta = betterproto.message_field(1)
@dataclass(eq=False, repr=False)
class TallyParams(betterproto.Message):
"""
TallyParams defines the params for tallying votes on governance proposals.
"""
# Minimum percentage of total stake needed to vote for a result to be
# considered valid.
quorum: bytes = betterproto.bytes_field(1)
# Minimum proportion of Yes votes for proposal to pass. Default value: 0.5.
threshold: bytes = betterproto.bytes_field(2)
# Minimum value of Veto votes to Total votes ratio for proposal to be
# vetoed. Default value: 1/3.
veto_threshold: bytes = betterproto.bytes_field(3)
@dataclass(eq=False, repr=False)
class MsgSubmitProposal(betterproto.Message):
"""
MsgSubmitProposal defines an sdk.Msg type that supports submitting
arbitrary proposal Content.
"""
content: "betterproto_lib_google_protobuf.Any" = betterproto.message_field(1)
initial_deposit: List["__base_v1_beta1__.Coin"] = betterproto.message_field(2)
proposer: str = betterproto.string_field(3)
@dataclass(eq=False, repr=False)
class MsgSubmitProposalResponse(betterproto.Message):
"""
MsgSubmitProposalResponse defines the Msg/SubmitProposal response type.
"""
proposal_id: int = betterproto.uint64_field(1)
@dataclass(eq=False, repr=False)
class MsgVote(betterproto.Message):
"""MsgVote defines a message to cast a vote."""
proposal_id: int = betterproto.uint64_field(1)
voter: str = betterproto.string_field(2)
option: "VoteOption" = betterproto.enum_field(3)
@dataclass(eq=False, repr=False)
class MsgVoteResponse(betterproto.Message):
"""MsgVoteResponse defines the Msg/Vote response type."""
pass
@dataclass(eq=False, repr=False)
class MsgVoteWeighted(betterproto.Message):
"""MsgVoteWeighted defines a message to cast a vote."""
proposal_id: int = betterproto.uint64_field(1)
voter: str = betterproto.string_field(2)
options: List["WeightedVoteOption"] = betterproto.message_field(3)
@dataclass(eq=False, repr=False)
class MsgVoteWeightedResponse(betterproto.Message):
"""MsgVoteWeightedResponse defines the Msg/VoteWeighted response type."""
pass
@dataclass(eq=False, repr=False)
class MsgDeposit(betterproto.Message):
"""
MsgDeposit defines a message to submit a deposit to an existing proposal.
"""
proposal_id: int = betterproto.uint64_field(1)
depositor: str = betterproto.string_field(2)
amount: List["__base_v1_beta1__.Coin"] = betterproto.message_field(3)
@dataclass(eq=False, repr=False)
class MsgDepositResponse(betterproto.Message):
"""MsgDepositResponse defines the Msg/Deposit response type."""
pass
@dataclass(eq=False, repr=False)
class QueryProposalRequest(betterproto.Message):
"""
QueryProposalRequest is the request type for the Query/Proposal RPC method.
"""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
@dataclass(eq=False, repr=False)
class QueryProposalResponse(betterproto.Message):
"""
QueryProposalResponse is the response type for the Query/Proposal RPC
method.
"""
proposal: "Proposal" = betterproto.message_field(1)
@dataclass(eq=False, repr=False)
class QueryProposalsRequest(betterproto.Message):
"""
QueryProposalsRequest is the request type for the Query/Proposals RPC
method.
"""
# proposal_status defines the status of the proposals.
proposal_status: "ProposalStatus" = betterproto.enum_field(1)
# voter defines the voter address for the proposals.
voter: str = betterproto.string_field(2)
# depositor defines the deposit addresses from the proposals.
depositor: str = betterproto.string_field(3)
# pagination defines an optional pagination for the request.
pagination: "__base_query_v1_beta1__.PageRequest" = betterproto.message_field(4)
@dataclass(eq=False, repr=False)
class QueryProposalsResponse(betterproto.Message):
"""
QueryProposalsResponse is the response type for the Query/Proposals RPC
method.
"""
proposals: List["Proposal"] = betterproto.message_field(1)
# pagination defines the pagination in the response.
pagination: "__base_query_v1_beta1__.PageResponse" = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class QueryVoteRequest(betterproto.Message):
"""QueryVoteRequest is the request type for the Query/Vote RPC method."""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
# voter defines the oter address for the proposals.
voter: str = betterproto.string_field(2)
@dataclass(eq=False, repr=False)
class QueryVoteResponse(betterproto.Message):
"""
QueryVoteResponse is the response type for the Query/Vote RPC method.
"""
# vote defined the queried vote.
vote: "Vote" = betterproto.message_field(1)
@dataclass(eq=False, repr=False)
class QueryVotesRequest(betterproto.Message):
"""
QueryVotesRequest is the request type for the Query/Votes RPC method.
"""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
# pagination defines an optional pagination for the request.
pagination: "__base_query_v1_beta1__.PageRequest" = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class QueryVotesResponse(betterproto.Message):
"""
QueryVotesResponse is the response type for the Query/Votes RPC method.
"""
# votes defined the queried votes.
votes: List["Vote"] = betterproto.message_field(1)
# pagination defines the pagination in the response.
pagination: "__base_query_v1_beta1__.PageResponse" = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class QueryParamsRequest(betterproto.Message):
"""
QueryParamsRequest is the request type for the Query/Params RPC method.
"""
# params_type defines which parameters to query for, can be one of "voting",
# "tallying" or "deposit".
params_type: str = betterproto.string_field(1)
@dataclass(eq=False, repr=False)
class QueryParamsResponse(betterproto.Message):
"""
QueryParamsResponse is the response type for the Query/Params RPC method.
"""
# voting_params defines the parameters related to voting.
voting_params: "VotingParams" = betterproto.message_field(1)
# deposit_params defines the parameters related to deposit.
deposit_params: "DepositParams" = betterproto.message_field(2)
# tally_params defines the parameters related to tally.
tally_params: "TallyParams" = betterproto.message_field(3)
@dataclass(eq=False, repr=False)
class QueryDepositRequest(betterproto.Message):
"""
QueryDepositRequest is the request type for the Query/Deposit RPC method.
"""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
# depositor defines the deposit addresses from the proposals.
depositor: str = betterproto.string_field(2)
@dataclass(eq=False, repr=False)
class QueryDepositResponse(betterproto.Message):
"""
QueryDepositResponse is the response type for the Query/Deposit RPC method.
"""
# deposit defines the requested deposit.
deposit: "Deposit" = betterproto.message_field(1)
@dataclass(eq=False, repr=False)
class QueryDepositsRequest(betterproto.Message):
"""
QueryDepositsRequest is the request type for the Query/Deposits RPC method.
"""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
# pagination defines an optional pagination for the request.
pagination: "__base_query_v1_beta1__.PageRequest" = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class QueryDepositsResponse(betterproto.Message):
"""
QueryDepositsResponse is the response type for the Query/Deposits RPC
method.
"""
deposits: List["Deposit"] = betterproto.message_field(1)
# pagination defines the pagination in the response.
pagination: "__base_query_v1_beta1__.PageResponse" = betterproto.message_field(2)
@dataclass(eq=False, repr=False)
class QueryTallyResultRequest(betterproto.Message):
"""
QueryTallyResultRequest is the request type for the Query/Tally RPC method.
"""
# proposal_id defines the unique id of the proposal.
proposal_id: int = betterproto.uint64_field(1)
@dataclass(eq=False, repr=False)
class QueryTallyResultResponse(betterproto.Message):
"""
QueryTallyResultResponse is the response type for the Query/Tally RPC
method.
"""
# tally defines the requested tally.
tally: "TallyResult" = betterproto.message_field(1)
@dataclass(eq=False, repr=False)
class GenesisState(betterproto.Message):
"""GenesisState defines the gov module's genesis state."""
# starting_proposal_id is the ID of the starting proposal.
starting_proposal_id: int = betterproto.uint64_field(1)
# deposits defines all the deposits present at genesis.
deposits: List["Deposit"] = betterproto.message_field(2)
# votes defines all the votes present at genesis.
votes: List["Vote"] = betterproto.message_field(3)
# proposals defines all the proposals present at genesis.
proposals: List["Proposal"] = betterproto.message_field(4)
# params defines all the paramaters of related to deposit.
deposit_params: "DepositParams" = betterproto.message_field(5)
# params defines all the paramaters of related to voting.
voting_params: "VotingParams" = betterproto.message_field(6)
# params defines all the paramaters of related to tally.
tally_params: "TallyParams" = betterproto.message_field(7)
class MsgStub(betterproto.ServiceStub):
async def submit_proposal(
self,
*,
content: "betterproto_lib_google_protobuf.Any" = None,
initial_deposit: Optional[List["__base_v1_beta1__.Coin"]] = None,
proposer: str = ""
) -> "MsgSubmitProposalResponse":
initial_deposit = initial_deposit or []
request = MsgSubmitProposal()
if content is not None:
request.content = content
if initial_deposit is not None:
request.initial_deposit = initial_deposit
request.proposer = proposer
return await self._unary_unary(
"/cosmos.gov.v1beta1.Msg/SubmitProposal", request, MsgSubmitProposalResponse
)
async def vote(
self, *, proposal_id: int = 0, voter: str = "", option: "VoteOption" = 0
) -> "MsgVoteResponse":
request = MsgVote()
request.proposal_id = proposal_id
request.voter = voter
request.option = option
return await self._unary_unary(
"/cosmos.gov.v1beta1.Msg/Vote", request, MsgVoteResponse
)
async def vote_weighted(
self,
*,
proposal_id: int = 0,
voter: str = "",
options: Optional[List["WeightedVoteOption"]] = None
) -> "MsgVoteWeightedResponse":
options = options or []
request = MsgVoteWeighted()
request.proposal_id = proposal_id
request.voter = voter
if options is not None:
request.options = options
return await self._unary_unary(
"/cosmos.gov.v1beta1.Msg/VoteWeighted", request, MsgVoteWeightedResponse
)
async def deposit(
self,
*,
proposal_id: int = 0,
depositor: str = "",
amount: Optional[List["__base_v1_beta1__.Coin"]] = None
) -> "MsgDepositResponse":
amount = amount or []
request = MsgDeposit()
request.proposal_id = proposal_id
request.depositor = depositor
if amount is not None:
request.amount = amount
return await self._unary_unary(
"/cosmos.gov.v1beta1.Msg/Deposit", request, MsgDepositResponse
)
class QueryStub(betterproto.ServiceStub):
async def proposal(self, *, proposal_id: int = 0) -> "QueryProposalResponse":
request = QueryProposalRequest()
request.proposal_id = proposal_id
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Proposal", request, QueryProposalResponse
)
async def proposals(
self,
*,
proposal_status: "ProposalStatus" = 0,
voter: str = "",
depositor: str = "",
pagination: "__base_query_v1_beta1__.PageRequest" = None
) -> "QueryProposalsResponse":
request = QueryProposalsRequest()
request.proposal_status = proposal_status
request.voter = voter
request.depositor = depositor
if pagination is not None:
request.pagination = pagination
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Proposals", request, QueryProposalsResponse
)
async def vote(
self, *, proposal_id: int = 0, voter: str = ""
) -> "QueryVoteResponse":
request = QueryVoteRequest()
request.proposal_id = proposal_id
request.voter = voter
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Vote", request, QueryVoteResponse
)
async def votes(
self,
*,
proposal_id: int = 0,
pagination: "__base_query_v1_beta1__.PageRequest" = None
) -> "QueryVotesResponse":
request = QueryVotesRequest()
request.proposal_id = proposal_id
if pagination is not None:
request.pagination = pagination
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Votes", request, QueryVotesResponse
)
async def params(self, *, params_type: str = "") -> "QueryParamsResponse":
request = QueryParamsRequest()
request.params_type = params_type
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Params", request, QueryParamsResponse
)
async def deposit(
self, *, proposal_id: int = 0, depositor: str = ""
) -> "QueryDepositResponse":
request = QueryDepositRequest()
request.proposal_id = proposal_id
request.depositor = depositor
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Deposit", request, QueryDepositResponse
)
async def deposits(
self,
*,
proposal_id: int = 0,
pagination: "__base_query_v1_beta1__.PageRequest" = None
) -> "QueryDepositsResponse":
request = QueryDepositsRequest()
request.proposal_id = proposal_id
if pagination is not None:
request.pagination = pagination
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/Deposits", request, QueryDepositsResponse
)
async def tally_result(self, *, proposal_id: int = 0) -> "QueryTallyResultResponse":
request = QueryTallyResultRequest()
request.proposal_id = proposal_id
return await self._unary_unary(
"/cosmos.gov.v1beta1.Query/TallyResult", request, QueryTallyResultResponse
)
class MsgBase(ServiceBase):
async def submit_proposal(
self,
content: "betterproto_lib_google_protobuf.Any",
initial_deposit: Optional[List["__base_v1_beta1__.Coin"]],
proposer: str,
) -> "MsgSubmitProposalResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def vote(
self, proposal_id: int, voter: str, option: "VoteOption"
) -> "MsgVoteResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def vote_weighted(
self,
proposal_id: int,
voter: str,
options: Optional[List["WeightedVoteOption"]],
) -> "MsgVoteWeightedResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def deposit(
self,
proposal_id: int,
depositor: str,
amount: Optional[List["__base_v1_beta1__.Coin"]],
) -> "MsgDepositResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_submit_proposal(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"content": request.content,
"initial_deposit": request.initial_deposit,
"proposer": request.proposer,
}
response = await self.submit_proposal(**request_kwargs)
await stream.send_message(response)
async def __rpc_vote(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"voter": request.voter,
"option": request.option,
}
response = await self.vote(**request_kwargs)
await stream.send_message(response)
async def __rpc_vote_weighted(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"voter": request.voter,
"options": request.options,
}
response = await self.vote_weighted(**request_kwargs)
await stream.send_message(response)
async def __rpc_deposit(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"depositor": request.depositor,
"amount": request.amount,
}
response = await self.deposit(**request_kwargs)
await stream.send_message(response)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/cosmos.gov.v1beta1.Msg/SubmitProposal": grpclib.const.Handler(
self.__rpc_submit_proposal,
grpclib.const.Cardinality.UNARY_UNARY,
MsgSubmitProposal,
MsgSubmitProposalResponse,
),
"/cosmos.gov.v1beta1.Msg/Vote": grpclib.const.Handler(
self.__rpc_vote,
grpclib.const.Cardinality.UNARY_UNARY,
MsgVote,
MsgVoteResponse,
),
"/cosmos.gov.v1beta1.Msg/VoteWeighted": grpclib.const.Handler(
self.__rpc_vote_weighted,
grpclib.const.Cardinality.UNARY_UNARY,
MsgVoteWeighted,
MsgVoteWeightedResponse,
),
"/cosmos.gov.v1beta1.Msg/Deposit": grpclib.const.Handler(
self.__rpc_deposit,
grpclib.const.Cardinality.UNARY_UNARY,
MsgDeposit,
MsgDepositResponse,
),
}
class QueryBase(ServiceBase):
async def proposal(self, proposal_id: int) -> "QueryProposalResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def proposals(
self,
proposal_status: "ProposalStatus",
voter: str,
depositor: str,
pagination: "__base_query_v1_beta1__.PageRequest",
) -> "QueryProposalsResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def vote(self, proposal_id: int, voter: str) -> "QueryVoteResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def votes(
self, proposal_id: int, pagination: "__base_query_v1_beta1__.PageRequest"
) -> "QueryVotesResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def params(self, params_type: str) -> "QueryParamsResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def deposit(self, proposal_id: int, depositor: str) -> "QueryDepositResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def deposits(
self, proposal_id: int, pagination: "__base_query_v1_beta1__.PageRequest"
) -> "QueryDepositsResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def tally_result(self, proposal_id: int) -> "QueryTallyResultResponse":
raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
async def __rpc_proposal(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
}
response = await self.proposal(**request_kwargs)
await stream.send_message(response)
async def __rpc_proposals(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_status": request.proposal_status,
"voter": request.voter,
"depositor": request.depositor,
"pagination": request.pagination,
}
response = await self.proposals(**request_kwargs)
await stream.send_message(response)
async def __rpc_vote(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"voter": request.voter,
}
response = await self.vote(**request_kwargs)
await stream.send_message(response)
async def __rpc_votes(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"pagination": request.pagination,
}
response = await self.votes(**request_kwargs)
await stream.send_message(response)
async def __rpc_params(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"params_type": request.params_type,
}
response = await self.params(**request_kwargs)
await stream.send_message(response)
async def __rpc_deposit(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"depositor": request.depositor,
}
response = await self.deposit(**request_kwargs)
await stream.send_message(response)
async def __rpc_deposits(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
"pagination": request.pagination,
}
response = await self.deposits(**request_kwargs)
await stream.send_message(response)
async def __rpc_tally_result(self, stream: grpclib.server.Stream) -> None:
request = await stream.recv_message()
request_kwargs = {
"proposal_id": request.proposal_id,
}
response = await self.tally_result(**request_kwargs)
await stream.send_message(response)
def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
return {
"/cosmos.gov.v1beta1.Query/Proposal": grpclib.const.Handler(
self.__rpc_proposal,
grpclib.const.Cardinality.UNARY_UNARY,
QueryProposalRequest,
QueryProposalResponse,
),
"/cosmos.gov.v1beta1.Query/Proposals": grpclib.const.Handler(
self.__rpc_proposals,
grpclib.const.Cardinality.UNARY_UNARY,
QueryProposalsRequest,
QueryProposalsResponse,
),
"/cosmos.gov.v1beta1.Query/Vote": grpclib.const.Handler(
self.__rpc_vote,
grpclib.const.Cardinality.UNARY_UNARY,
QueryVoteRequest,
QueryVoteResponse,
),
"/cosmos.gov.v1beta1.Query/Votes": grpclib.const.Handler(
self.__rpc_votes,
grpclib.const.Cardinality.UNARY_UNARY,
QueryVotesRequest,
QueryVotesResponse,
),
"/cosmos.gov.v1beta1.Query/Params": grpclib.const.Handler(
self.__rpc_params,
grpclib.const.Cardinality.UNARY_UNARY,
QueryParamsRequest,
QueryParamsResponse,
),
"/cosmos.gov.v1beta1.Query/Deposit": grpclib.const.Handler(
self.__rpc_deposit,
grpclib.const.Cardinality.UNARY_UNARY,
QueryDepositRequest,
QueryDepositResponse,
),
"/cosmos.gov.v1beta1.Query/Deposits": grpclib.const.Handler(
self.__rpc_deposits,
grpclib.const.Cardinality.UNARY_UNARY,
QueryDepositsRequest,
QueryDepositsResponse,
),
"/cosmos.gov.v1beta1.Query/TallyResult": grpclib.const.Handler(
self.__rpc_tally_result,
grpclib.const.Cardinality.UNARY_UNARY,
QueryTallyResultRequest,
QueryTallyResultResponse,
),
}
from ...base import v1beta1 as __base_v1_beta1__
from ...base.query import v1beta1 as __base_query_v1_beta1__
import betterproto.lib.google.protobuf as betterproto_lib_google_protobuf