Chain Modules

class GrpcOptions(TypedDict, total=False):
  """Options for constructing a gRPC transport."""

  port: int
  """gRPC endpoint port."""
  ssl: bool
  """Use TLS for the gRPC channel."""

class Modules(GrpcRouter):
  """Composed dYdX Chain module groups."""

  auth: Auth
  bank: Bank
  tx: Tx
  tendermint: Tendermint
  staking: Staking
  distribution: Distribution
  subaccounts: Subaccounts
  clob: Clob
  prices: Prices
  perpetuals: Perpetuals
  assets: Assets
  feetiers: Feetiers
  rewards: Rewards
  affiliates: Affiliates
  revshare: Revshare

  @classmethod
  def new(cls, host: str, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for a custom gRPC endpoint."""
    return cls(client=GrpcClient(host=host, **kwargs))

  @classmethod
  def from_client(cls, client: GrpcClient) -> Self:
    """Create module groups from an existing gRPC transport."""
    return cls(client=client)

  @classmethod
  def oegs(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the OEGS mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_OEGS_HOST, **kwargs)

  @classmethod
  def polkachu(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the first Polkachu mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_POLKACHU_1_HOST, **kwargs)

  @classmethod
  def polkachu_2(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the second Polkachu mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_POLKACHU_2_HOST, **kwargs)

  @classmethod
  def polkachu_3(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the third Polkachu mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_POLKACHU_3_HOST, **kwargs)

  @classmethod
  def kingnodes(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the KingNodes mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_KINGNODES_HOST, **kwargs)

  @classmethod
  def enigma(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the Enigma mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_ENIGMA_HOST, **kwargs)

  @classmethod
  def polkachu_archive(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the Polkachu archive mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_POLKACHU_ARCHIVE_1_HOST, **kwargs)

  @classmethod
  def kingnodes_archive(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the KingNodes archive mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_KINGNODES_ARCHIVE_HOST, **kwargs)

  @classmethod
  def enigma_archive(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the Enigma archive mainnet gRPC endpoint."""
    return cls.new(DYDX_GRPC_ENIGMA_ARCHIVE_HOST, **kwargs)

  @classmethod
  def testnet_oegs(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the OEGS testnet gRPC endpoint."""
    return cls.new(DYDX_TESTNET_GRPC_OEGS_HOST, **kwargs)

  @classmethod
  def testnet_kingnodes(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the KingNodes testnet gRPC endpoint."""
    return cls.new(DYDX_TESTNET_GRPC_KINGNODES_HOST, **kwargs)

  @classmethod
  def testnet_polkachu(cls, **kwargs: Unpack[GrpcOptions]) -> Self:
    """Create module groups for the Polkachu plaintext testnet gRPC endpoint."""
    port = kwargs.get('port', 23890)
    ssl = kwargs.get('ssl', False)
    return cls.new(DYDX_TESTNET_GRPC_POLKACHU_HOST, port=port, ssl=ssl)

class ModulesOptions(TypedDict, total=False):
  """Options shared by gRPC module constructors."""

  client: GrpcClient
  """Existing gRPC transport shared by module adapters."""

class Auth(Account, AccountInfo, Accounts, GrpcEndpoint):
  """Cosmos auth query group."""

class Account(GrpcEndpoint):
  """Auth account endpoint."""

  async def account(self, address: str) -> auth_proto.QueryAccountResponse:
    """Query an account by address."""
    return await auth_proto.QueryStub(self.channel).account(auth_proto.QueryAccountRequest(address=address))

class AccountInfo(GrpcEndpoint):
  """Auth account info endpoint."""

  async def account_info(self, address: str) -> auth_proto.QueryAccountInfoResponse:
    """Query account sequence and number information."""
    return await auth_proto.QueryStub(self.channel).account_info(auth_proto.QueryAccountInfoRequest(address=address))

class Accounts(GrpcEndpoint):
  """Auth accounts endpoint."""

  async def accounts(self, *, pagination: query_proto.PageRequest | None = None) -> auth_proto.QueryAccountsResponse:
    """Query chain accounts."""
    request = auth_proto.QueryAccountsRequest(pagination=pagination)
    return await auth_proto.QueryStub(self.channel).accounts(request)

  def accounts_paged(self, *, limit: int | None = None) -> PaginatedResponse[Any, bytes]:
    """Page through chain accounts.

    Args:
      limit: Optional maximum number of accounts per page.

    Returns:
      A paginated response yielding account payload pages.
    """
    async def next(key: bytes) -> tuple[list[Any], bytes | None]:
      """Fetch the next account page."""
      response = await self.accounts(pagination=page_request(key, limit=limit))
      return response.accounts, next_key(response)

    return PaginatedResponse(b'', next)

class Bank(
  AllBalances,
  Balance,
  DenomMetadata,
  DenomMetadataByQueryString,
  DenomsMetadata,
  Params,
  SpendableBalances,
  Supply,
  TotalSupply,
  GrpcEndpoint,
):
  """Cosmos bank query group."""

class AllBalances(GrpcEndpoint):
  """Bank all balances endpoint."""

  async def all_balances(
    self, address: str, *, pagination: query_proto.PageRequest | None = None, resolve_denom: bool = False,
  ) -> bank_proto.QueryAllBalancesResponse:
    """Query all balances for an account address."""
    request = bank_proto.QueryAllBalancesRequest(
      address=address,
      pagination=pagination,
      resolve_denom=resolve_denom,
    )
    return await bank_proto.QueryStub(self.channel).all_balances(request)

  def all_balances_paged(
    self, address: str, *, limit: int | None = None, resolve_denom: bool = False,
  ) -> PaginatedResponse[coin_proto.Coin, bytes]:
    """Page through all balances for an account address.

    Args:
      address: Account address to query.
      limit: Optional maximum number of balances per page.
      resolve_denom: Resolve denoms into human-readable metadata when supported.

    Returns:
      A paginated response yielding balance pages.
    """
    async def next(key: bytes) -> tuple[list[coin_proto.Coin], bytes | None]:
      """Fetch the next balance page."""
      response = await self.all_balances(
        address,
        pagination=page_request(key, limit=limit),
        resolve_denom=resolve_denom,
      )
      return response.balances, next_key(response)

    return PaginatedResponse(b'', next)

class Balance(GrpcEndpoint):
  """Bank balance endpoint."""

  async def balance(self, *, address: str, denom: str) -> bank_proto.QueryBalanceResponse:
    """Query one denom balance for an account address."""
    return await bank_proto.QueryStub(self.channel).balance(
      bank_proto.QueryBalanceRequest(address=address, denom=denom)
    )

class DenomMetadata(GrpcEndpoint):
  """Bank denom metadata endpoint."""

  async def denom_metadata(self, denom: str) -> bank_proto.QueryDenomMetadataResponse:
    """Query metadata for one denom.

    Args:
      denom: Base or display denomination to query.

    Returns:
      Metadata describing the denom display units and exponent.
    """
    request = bank_proto.QueryDenomMetadataRequest(denom=denom)
    return await bank_proto.QueryStub(self.channel).denom_metadata(request)

class DenomMetadataByQueryString(GrpcEndpoint):
  """Bank denom metadata by query string endpoint."""

  async def denom_metadata_by_query_string(
    self, denom: str,
  ) -> bank_proto.QueryDenomMetadataByQueryStringResponse:
    """Query metadata for one denom through the alternate query-string RPC.

    Args:
      denom: Base or display denomination to query.

    Returns:
      Metadata describing the denom display units and exponent.
    """
    request = bank_proto.QueryDenomMetadataByQueryStringRequest(denom=denom)
    return await bank_proto.QueryStub(self.channel).denom_metadata_by_query_string(request)

class DenomsMetadata(GrpcEndpoint):
  """Bank denoms metadata endpoint."""

  async def denoms_metadata(
    self, *, pagination: query_proto.PageRequest | None = None,
  ) -> bank_proto.QueryDenomsMetadataResponse:
    """Query metadata for all registered denoms.

    Args:
      pagination: Optional page controls for large metadata registries.

    Returns:
      Metadata entries describing denom display units and exponents.
    """
    request = bank_proto.QueryDenomsMetadataRequest(pagination=pagination)
    return await bank_proto.QueryStub(self.channel).denoms_metadata(request)

  def denoms_metadata_paged(
    self, *, limit: int | None = None,
  ) -> PaginatedResponse[bank_proto.Metadata, bytes]:
    """Page through metadata for all registered denoms.

    Args:
      limit: Optional maximum number of metadata entries per page.

    Returns:
      A paginated response yielding denom metadata pages.
    """
    async def next(key: bytes) -> tuple[list[bank_proto.Metadata], bytes | None]:
      """Fetch the next denom metadata page."""
      response = await self.denoms_metadata(pagination=page_request(key, limit=limit))
      return response.metadatas, next_key(response)

    return PaginatedResponse(b'', next)

class Params(GrpcEndpoint):
  """Bank params endpoint."""

  async def params(self) -> bank_proto.QueryParamsResponse:
    """Query bank module parameters."""
    return await bank_proto.QueryStub(self.channel).params(bank_proto.QueryParamsRequest())

class SpendableBalances(GrpcEndpoint):
  """Bank spendable balances endpoint."""

  async def spendable_balances(
    self, address: str, *, pagination: query_proto.PageRequest | None = None,
  ) -> bank_proto.QuerySpendableBalancesResponse:
    """Query spendable balances for an account address."""
    request = bank_proto.QuerySpendableBalancesRequest(address=address, pagination=pagination)
    return await bank_proto.QueryStub(self.channel).spendable_balances(request)

  def spendable_balances_paged(
    self, address: str, *, limit: int | None = None,
  ) -> PaginatedResponse[coin_proto.Coin, bytes]:
    """Page through spendable balances for an account address.

    Args:
      address: Account address to query.
      limit: Optional maximum number of balances per page.

    Returns:
      A paginated response yielding spendable balance pages.
    """
    async def next(key: bytes) -> tuple[list[coin_proto.Coin], bytes | None]:
      """Fetch the next spendable balance page."""
      response = await self.spendable_balances(
        address,
        pagination=page_request(key, limit=limit),
      )
      return response.balances, next_key(response)

    return PaginatedResponse(b'', next)

class Supply(GrpcEndpoint):
  """Bank denom supply endpoint."""

  async def supply_of(self, denom: str) -> bank_proto.QuerySupplyOfResponse:
    """Query supply for one denom."""
    return await bank_proto.QueryStub(self.channel).supply_of(bank_proto.QuerySupplyOfRequest(denom=denom))

class TotalSupply(GrpcEndpoint):
  """Bank total supply endpoint."""

  async def total_supply(
    self, *, pagination: query_proto.PageRequest | None = None,
  ) -> bank_proto.QueryTotalSupplyResponse:
    """Query total token supply."""
    request = bank_proto.QueryTotalSupplyRequest(pagination=pagination)
    return await bank_proto.QueryStub(self.channel).total_supply(request)

  def total_supply_paged(self, *, limit: int | None = None) -> PaginatedResponse[coin_proto.Coin, bytes]:
    """Page through total token supply.

    Args:
      limit: Optional maximum number of supply entries per page.

    Returns:
      A paginated response yielding supply pages.
    """
    async def next(key: bytes) -> tuple[list[coin_proto.Coin], bytes | None]:
      """Fetch the next supply page."""
      response = await self.total_supply(pagination=page_request(key, limit=limit))
      return response.supply, next_key(response)

    return PaginatedResponse(b'', next)

class Tx(Broadcast, GetTx, GetTxsEvent, Simulate, GrpcEndpoint):
  """Cosmos transaction service group."""

class Broadcast(GrpcEndpoint):
  """Transaction broadcast endpoint."""

  async def broadcast(
    self,
    tx_bytes: bytes,
    *,
    mode: tx_proto.BroadcastMode = tx_proto.BroadcastMode.SYNC,
  ) -> tx_proto.BroadcastTxResponse:
    """Broadcast encoded transaction bytes."""
    request = tx_proto.BroadcastTxRequest(tx_bytes=tx_bytes, mode=mode)
    return await tx_proto.ServiceStub(self.channel).broadcast_tx(request)

class GetTx(GrpcEndpoint):
  """Transaction lookup endpoint."""

  async def get_tx(self, hash: str) -> tx_proto.GetTxResponse:
    """Query a transaction by hash."""
    return await tx_proto.ServiceStub(self.channel).get_tx(tx_proto.GetTxRequest(hash=hash))

class GetTxsEvent(GrpcEndpoint):
  """Transaction event search endpoint."""

  async def get_txs_event(
    self,
    query: str | None = None, *,
    events: list[str] | None = None,
    pagination: query_proto.PageRequest | None = None,
    order_by: OrderBy | None = None,
    page: int | None = None,
    limit: int | None = None,
  ) -> tx_proto.GetTxsEventResponse:
    """Query transactions by event expression."""
    request = tx_proto.GetTxsEventRequest()
    if pagination is not None:
      request.pagination = pagination
    if order_by is not None:
      request.order_by = order_by
    if page is not None:
      request.page = page
    if limit is not None:
      request.limit = limit
    if query is not None:
      request.query = query
    if events:
      request.events.extend(events)
    return await tx_proto.ServiceStub(self.channel).get_txs_event(request)

  def get_txs_event_paged(
    self,
    query: str | None = None, *,
    events: list[str] | None = None,
    order_by: OrderBy | None = None,
    limit: int | None = None,
  ) -> PaginatedResponse[abci_proto.TxResponse, bytes]:
    """Page through transactions matching event filters.

    Args:
      query: Optional event query expression.
      events: Optional event filters for legacy query semantics.
      order_by: Optional result ordering.
      limit: Optional maximum number of transactions per page.

    Returns:
      A paginated response yielding transaction response pages.
    """
    async def next(key: bytes) -> tuple[list[abci_proto.TxResponse], bytes | None]:
      """Fetch the next transaction event page."""
      response = await self.get_txs_event(
        query,
        events=events,
        pagination=page_request(key, limit=limit),
        order_by=order_by,
      )
      return response.tx_responses, next_key(response)

    return PaginatedResponse(b'', next)