from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Callable, Literal, overload
import requests
from typing_extensions import Concatenate, ParamSpec
from evmos.eip712 import (
EIPToSign,
create_eip712,
generate_fee,
generate_message,
generate_message_with_multiple_transactions,
generate_types,
)
from evmos.proto import (
MessageGenerated,
create_transaction,
create_transaction_with_multiple_messages,
)
from evmos.proto.transactions import TxGeneratedBase as TxGeneratedBase
from evmos.provider import generate_endpoint_account
from evmos.utils.doc import _inherit
[docs]
@dataclass
class Fee:
"""Fee for message."""
amount: str
"""Fee amount (stringified number, like '1000')."""
denom: str
"""Denomination."""
gas: str
"""Gas price."""
[docs]
@dataclass
class Sender:
"""Message sender."""
account_address: str
"""Account address (bech32, ``evmos1...``)."""
sequence: int = 0
"""Account nonce - amount of previously sent transactions."""
account_number: int = 0
"""Internal account number."""
pubkey: str = ""
"""Account public key."""
[docs]
def update_from_chain(
self, url: str = "http://127.0.0.1:1317", timeout: int | None = 5
) -> None:
"""Set `sequence`, `account_number` and possibly `pubkey` from API response."""
response = requests.get(
f"{url}{generate_endpoint_account(self.account_address)}", timeout=timeout
)
resp = response.json()
self.sequence = int(resp["account"]["base_account"]["sequence"])
self.account_number = int(resp["account"]["base_account"]["account_number"])
if not self.pubkey:
self.pubkey = resp["account"]["base_account"]["pub_key"]["key"]
[docs]
@dataclass
class Chain:
"""Chain definition."""
chain_id: int
"""Main chain ID."""
cosmos_chain_id: str
"""Cosmos chain ID."""
[docs]
@dataclass
class TxGenerated(TxGeneratedBase):
"""Transaction generated by this library (with EIP to sign)."""
eip_to_sign: EIPToSign
"""EIP message to sign for EIP-712 transactions."""
_P = ParamSpec("_P")
[docs]
def to_generated_base(
func: Callable[Concatenate[str, _P], MessageGenerated[Any]],
) -> Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGeneratedBase]:
"""Wrap function returning message with transaction base."""
# Not using functools.wraps, because signature is altered
@_inherit(func)
def inner(
chain: Chain,
sender: Sender,
fee: Fee,
memo: str,
/,
*args: _P.args,
**kwargs: _P.kwargs,
) -> TxGeneratedBase:
msg_cosmos = func(sender.account_address, *args, **kwargs)
tx = create_transaction(
msg_cosmos,
memo,
fee.amount,
fee.denom,
int(fee.gas),
"ethsecp256",
sender.pubkey,
sender.sequence,
sender.account_number,
chain.cosmos_chain_id,
)
return TxGeneratedBase(
sign_direct=tx.sign_direct,
legacy_amino=tx.legacy_amino,
)
return inner
@overload
def to_generated(
types_def: dict[str, Any], *, proto: Literal[True], many: Literal[True]
) -> Callable[
[
Callable[
Concatenate[str, _P],
tuple[Sequence[Mapping[str, Any]], Sequence[MessageGenerated[Any]]],
]
],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]: ...
@overload
def to_generated(
types_def: dict[str, Any], *, proto: Literal[False] = ..., many: Literal[True]
) -> Callable[
[Callable[_P, tuple[Sequence[Mapping[str, Any]], Sequence[MessageGenerated[Any]]]]],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]: ...
@overload
def to_generated(
types_def: dict[str, Any], *, proto: Literal[True], many: Literal[False] = ...
) -> Callable[
[Callable[Concatenate[str, _P], tuple[Mapping[str, Any], MessageGenerated[Any]]]],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]: ...
@overload
def to_generated(
types_def: dict[str, Any],
*,
proto: Literal[False] = ...,
many: Literal[False] = ...,
) -> Callable[
[Callable[_P, tuple[Mapping[str, Any], MessageGenerated[Any]]]],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]: ...
[docs]
def to_generated(
types_def: dict[str, Any], *, proto: bool = False, many: bool = False
) -> (
Callable[
[Callable[_P, tuple[Mapping[str, Any], MessageGenerated[Any]]]],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]
| Callable[
[
Callable[
Concatenate[str, _P], tuple[Mapping[str, Any], MessageGenerated[Any]]
]
],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]
| Callable[
[
Callable[
_P, tuple[Sequence[Mapping[str, Any]], Sequence[MessageGenerated[Any]]]
]
],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]
| Callable[
[
Callable[
Concatenate[str, _P],
tuple[Sequence[Mapping[str, Any]], Sequence[MessageGenerated[Any]]],
]
],
Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated],
]
):
"""Wrap function returning message with transaction."""
def _inner(
chain: Chain,
sender: Sender,
fee: Fee,
memo: str,
msg: Mapping[str, Any] | Sequence[Mapping[str, Any]],
msg_cosmos: MessageGenerated[Any] | Sequence[MessageGenerated[Any]],
) -> TxGenerated:
# EIP712
fee_object = generate_fee(
fee.amount,
fee.denom,
fee.gas,
sender.account_address,
)
types = generate_types(types_def)
# No, I won't make it even more ugly with further typing
messages_factory = (
generate_message_with_multiple_transactions if many else generate_message
)
messages = messages_factory(
str(sender.account_number),
str(sender.sequence),
chain.cosmos_chain_id,
memo,
fee_object,
msg, # type: ignore[arg-type]
)
eip_to_sign = create_eip712(types, chain.chain_id, messages)
# Cosmos
tx_factory = (
create_transaction_with_multiple_messages if many else create_transaction
)
tx = tx_factory(
msg_cosmos,
memo,
fee.amount,
fee.denom,
int(fee.gas),
"ethsecp256",
sender.pubkey,
sender.sequence,
sender.account_number,
chain.cosmos_chain_id,
)
return TxGenerated(
sign_direct=tx.sign_direct,
legacy_amino=tx.legacy_amino,
eip_to_sign=eip_to_sign,
)
if proto:
def decorator(
func: Callable[
Concatenate[str, _P],
tuple[
Mapping[str, Any] | Sequence[Mapping[str, Any]],
MessageGenerated[Any] | Sequence[MessageGenerated[Any]],
],
],
) -> Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated]:
@_inherit(func)
def inner(
chain: Chain,
sender: Sender,
fee: Fee,
memo: str,
/,
*args: _P.args,
**kwargs: _P.kwargs,
) -> TxGenerated:
msg, msg_cosmos = func(sender.account_address, *args, **kwargs)
return _inner(chain, sender, fee, memo, msg, msg_cosmos)
return inner
return decorator
def decorator2(
func: Callable[
_P,
tuple[
Mapping[str, Any] | Sequence[Mapping[str, Any]],
MessageGenerated[Any] | Sequence[MessageGenerated[Any]],
],
],
) -> Callable[Concatenate[Chain, Sender, Fee, str, _P], TxGenerated]:
@_inherit(func)
def inner(
chain: Chain,
sender: Sender,
fee: Fee,
memo: str,
/,
*args: _P.args,
**kwargs: _P.kwargs,
) -> TxGenerated:
msg, msg_cosmos = func(*args, **kwargs)
return _inner(chain, sender, fee, memo, msg, msg_cosmos)
return inner
return decorator2