from typing import Sequence
from multiversx_sdk.abi import Serializer
from multiversx_sdk.abi.biguint_value import BigUIntValue
from multiversx_sdk.abi.string_value import StringValue
from multiversx_sdk.builders.transaction_builder import TransactionBuilder
from multiversx_sdk.core import Address, Transaction
from multiversx_sdk.core.constants import DELEGATION_MANAGER_SC_ADDRESS_HEX
from multiversx_sdk.core.transactions_factory_config import \
TransactionsFactoryConfig
from multiversx_sdk.delegation.errors import ListsLengthMismatchError
from multiversx_sdk.wallet.validator_keys import ValidatorPublicKey
[docs]
class DelegationTransactionsFactory:
def __init__(self, config: TransactionsFactoryConfig) -> None:
self.config = config
self.serializer = Serializer()
[docs]
def create_transaction_for_new_delegation_contract(self,
sender: Address,
total_delegation_cap: int,
service_fee: int,
amount: int) -> Transaction:
parts = ["createNewDelegationContract"]
serialized_parts = self.serializer.serialize_to_parts([
BigUIntValue(total_delegation_cap),
BigUIntValue(service_fee)
])
parts.extend([part.hex() for part in serialized_parts])
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=Address.new_from_hex(DELEGATION_MANAGER_SC_ADDRESS_HEX),
data_parts=parts,
gas_limit=self.config.gas_limit_create_delegation_contract + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True,
amount=amount
).build()
return transaction
[docs]
def create_transaction_for_adding_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey],
signed_messages: Sequence[bytes]) -> Transaction:
if len(public_keys) != len(signed_messages):
raise ListsLengthMismatchError("The number of public keys should match the number of signed messages")
parts = ["addNodes"]
for i in range(len(public_keys)):
parts.append(public_keys[i].hex())
parts.append(signed_messages[i].hex())
num_nodes = len(public_keys)
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self._compute_execution_gas_limit_for_nodes_management(num_nodes),
add_data_movement_gas=True
).build()
return transaction
def _compute_execution_gas_limit_for_nodes_management(self, num_nodes: int) -> int:
return self.config.gas_limit_delegation_operations + num_nodes * self.config.additional_gas_limit_per_validator_node
[docs]
def create_transaction_for_removing_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey]) -> Transaction:
num_nodes = len(public_keys)
parts = ["removeNodes"]
for public_key in public_keys:
parts.append(public_key.hex())
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self._compute_execution_gas_limit_for_nodes_management(num_nodes),
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_staking_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey]) -> Transaction:
num_nodes = len(public_keys)
parts = ["stakeNodes"]
for public_key in public_keys:
parts.append(public_key.hex())
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.gas_limit_stake +
num_nodes * self.config.additional_gas_limit_per_validator_node,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_unbonding_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey]) -> Transaction:
num_nodes = len(public_keys)
parts = ["unBondNodes"]
for public_key in public_keys:
parts.append(public_key.hex())
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.gas_limit_unbond +
num_nodes * self.config.additional_gas_limit_per_validator_node,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_unstaking_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey]) -> Transaction:
num_nodes = len(public_keys)
parts = ["unStakeNodes"]
for public_key in public_keys:
parts.append(public_key.hex())
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.gas_limit_unstake +
num_nodes * self.config.additional_gas_limit_per_validator_node,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_unjailing_nodes(self,
sender: Address,
delegation_contract: Address,
public_keys: Sequence[ValidatorPublicKey],
amount: int) -> Transaction:
num_nodes = len(public_keys)
parts = ["unJailNodes"]
for public_key in public_keys:
parts.append(public_key.hex())
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self._compute_execution_gas_limit_for_nodes_management(num_nodes),
add_data_movement_gas=True,
amount=amount
).build()
return transaction
[docs]
def create_transaction_for_changing_service_fee(self,
sender: Address,
delegation_contract: Address,
service_fee: int) -> Transaction:
parts = [
"changeServiceFee",
self.serializer.serialize([BigUIntValue(service_fee)])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_modifying_delegation_cap(self,
sender: Address,
delegation_contract: Address,
delegation_cap: int) -> Transaction:
parts = [
"modifyTotalDelegationCap",
self.serializer.serialize([BigUIntValue(delegation_cap)])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_setting_automatic_activation(self,
sender: Address,
delegation_contract: Address) -> Transaction:
parts = [
"setAutomaticActivation",
self.serializer.serialize([StringValue('true')])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_unsetting_automatic_activation(self,
sender: Address,
delegation_contract: Address) -> Transaction:
parts = [
"setAutomaticActivation",
self.serializer.serialize([StringValue('false')])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_setting_cap_check_on_redelegate_rewards(self,
sender: Address,
delegation_contract: Address) -> Transaction:
parts = [
"setCheckCapOnReDelegateRewards",
self.serializer.serialize([StringValue('true')])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_unsetting_cap_check_on_redelegate_rewards(self,
sender: Address,
delegation_contract: Address) -> Transaction:
parts = [
"setCheckCapOnReDelegateRewards",
self.serializer.serialize([StringValue('false')])
]
transaction = TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=parts,
gas_limit=self.config.gas_limit_delegation_operations + self.config.additional_gas_for_delegation_operations,
add_data_movement_gas=True
).build()
return transaction
[docs]
def create_transaction_for_delegating(self,
sender: Address,
delegation_contract: Address,
amount: int) -> Transaction:
return TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=["delegate"],
gas_limit=12000000,
add_data_movement_gas=False,
amount=amount
).build()
[docs]
def create_transaction_for_claiming_rewards(self,
sender: Address,
delegation_contract: Address) -> Transaction:
return TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=["claimRewards"],
gas_limit=6000000,
add_data_movement_gas=False,
amount=0
).build()
[docs]
def create_transaction_for_redelegating_rewards(self,
sender: Address,
delegation_contract: Address) -> Transaction:
return TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=["reDelegateRewards"],
gas_limit=12000000,
add_data_movement_gas=False,
amount=0
).build()
[docs]
def create_transaction_for_undelegating(self,
sender: Address,
delegation_contract: Address,
amount: int) -> Transaction:
return TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=["unDelegate", self.serializer.serialize([BigUIntValue(amount)])],
gas_limit=12000000,
add_data_movement_gas=False,
amount=0
).build()
[docs]
def create_transaction_for_withdrawing(self,
sender: Address,
delegation_contract: Address) -> Transaction:
return TransactionBuilder(
config=self.config,
sender=sender,
receiver=delegation_contract,
data_parts=["withdraw"],
gas_limit=12000000,
add_data_movement_gas=False,
amount=0
).build()