Source code for pypergraph.keyring.wallets.single_account_wallet

import re
from typing import Optional, List, Dict, Any, Union

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from pydantic import Field, model_serializer, model_validator, BaseModel

from pypergraph.core import KeyringAssetType, KeyringWalletType, NetworkId

from .shared import sid_manager
from ..accounts.dag_account import DagAccount
from ..accounts.eth_account import EthAccount
from ..keyrings.simple_keyring import SimpleKeyring



[docs]
class SingleAccountWallet(BaseModel):
    type: str = Field(default=KeyringWalletType.SingleAccountWallet.value)
    id: str = Field(default=None)
    supported_assets: List = Field(default_factory=list)
    label: Optional[str] = Field(default=None, max_length=12)
    keyring: Optional[SimpleKeyring] = Field(default=None)
    network: Optional[str] = Field(default=None)


[docs]
    @model_validator(mode="after")
    def compute_id(self):
        """Automatically computes the id based on injected SID value."""
        self.id = sid_manager.next_sid(self.type)
        return self



[docs]
    @model_serializer
    def model_serialize(self) -> Dict[str, Any]:
        return {
            "type": self.type,
            "label": self.label,
            "network": self.network,
            "secret": self.export_secret_key(),
        }



[docs]
    def create(self, network: str, label: str, private_key: str = None):
        """
        Initiates the creation of a new single key wallet.

        :param network: "Constellation" or "Ethereum".
        :param private_key: Optional, the private key to create account for. Leaving empty will create a new account from new private key.
        :param label: The name of the wallet.
        """

        private_key = (
            private_key
            or ec.generate_private_key(curve=ec.SECP256K1(), backend=default_backend())
            .private_numbers()
            .private_value.to_bytes(32, byteorder="big")
            .hex()
        )
        valid = re.fullmatch(r"^[a-fA-F0-9]{64}$", private_key)
        if not valid:
            ValueError("SingleAccountWallet :: Private key is invalid.")
        self.deserialize(label=label, network=network, secret=private_key)



[docs]
    def set_label(self, label: str):
        """
        Set the name of the wallet.

        :param label: The wallet name.
        """
        if not label:
            raise ValueError("SingleAccountWallet :: No label set.")
        self.label = label



[docs]
    def get_label(self) -> str:
        """
        Get the name of the wallet.

        :return: The wallet name.
        """
        return self.label



[docs]
    def get_network(self) -> str:
        return self.network



[docs]
    def get_state(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "type": self.type,
            "label": self.label,
            "supported_assets": self.supported_assets,
            "accounts": [
                {
                    "address": a.get_address(),
                    "network": a.get_network_id(),
                    "tokens": a.get_tokens(),
                }
                for a in self.get_accounts()
            ],
        }



[docs]
    def deserialize(self, network: str, label: str, secret: str):
        self.set_label(label)
        self.network = network or NetworkId.Constellation.value
        self.keyring = SimpleKeyring()

        self.keyring.deserialize(
            network=self.network, accounts=[{"private_key": secret}]
        )

        if self.network == NetworkId.Ethereum.value:
            self.supported_assets.append(KeyringAssetType.ETH.value)
            self.supported_assets.append(KeyringAssetType.ERC20.value)

        elif self.network == NetworkId.Constellation.value:
            self.supported_assets.append(KeyringAssetType.DAG.value)



[docs]
    @staticmethod
    def import_account():
        """Not supported for SingleAccountWallet."""
        raise ValueError(
            "SingleAccountWallet :: does not support importing of account."
        )



[docs]
    def get_accounts(self) -> List[Union[DagAccount, EthAccount]]:
        return self.keyring.get_accounts()



[docs]
    def get_account_by_address(self, address: str) -> Union[DagAccount, EthAccount]:
        """
        Get the account matching a specific address.

        :param address: The account address.
        :return: The account matching the address.
        """
        return self.keyring.get_account_by_address(address)



[docs]
    def remove_account(self, account):
        """Not supported by SAW."""
        raise ValueError("SingleChainWallet :: Does not allow removing accounts.")



[docs]
    def export_secret_key(self) -> str:
        """
        Get the privat key.

        :return: Private key in hexadecimal string format.
        """
        return (
            self.keyring.get_accounts()[0]
            .wallet.private_numbers()
            .private_value.to_bytes(32, "big")
            .hex()
        )



[docs]
    @staticmethod
    def reset_sid():
        sid_manager.reset_sid()