xrpl-dev-portal/_code-samples/vaults/py/vault_setup.py

import asyncio
import json
import sys

from xrpl.asyncio.clients import AsyncWebsocketClient
from xrpl.asyncio.transaction import submit_and_wait
from xrpl.asyncio.wallet import generate_faucet_wallet
from xrpl.models import (
    Batch, BatchFlag, CredentialAccept, CredentialCreate, MPTokenAuthorize,
    MPTokenIssuanceCreate, MPTokenIssuanceCreateFlag, Payment,
    PermissionedDomainSet, VaultDeposit
)
from xrpl.models.requests import AccountObjects
from xrpl.models.transactions.deposit_preauth import Credential
from xrpl.models.transactions.vault_create import (
    VaultCreate, VaultCreateFlag, WithdrawalPolicy
)
from xrpl.utils import encode_mptoken_metadata, str_to_hex


async def main():
    # Setup script for vault tutorials
    print("Setting up tutorial: 0/5", end="\r")

    async with AsyncWebsocketClient("wss://s.devnet.rippletest.net:51233") as client:
        # Create and fund all wallets concurrently
        mpt_issuer, domain_owner, depositor, vault_owner = await asyncio.gather(
            generate_faucet_wallet(client),
            generate_faucet_wallet(client),
            generate_faucet_wallet(client),
            generate_faucet_wallet(client),
        )

        # Step 1: Create MPT issuance, permissioned domain, and credentials in parallel
        print("Setting up tutorial: 1/5", end="\r")

        cred_type = "VaultAccess"
        mpt_create_result, _ = await asyncio.gather(
            submit_and_wait(
                MPTokenIssuanceCreate(
                    account=mpt_issuer.address,
                    flags=(
                        MPTokenIssuanceCreateFlag.TF_MPT_CAN_TRANSFER |
                        MPTokenIssuanceCreateFlag.TF_MPT_CAN_LOCK
                    ),
                    asset_scale=2,
                    transfer_fee=0,
                    maximum_amount="1000000000000",
                    mptoken_metadata=encode_mptoken_metadata({
                        "ticker": "USTST",
                        "name": "USTST Stablecoin",
                        "desc": "A test stablecoin token",
                        "icon": "example.org/ustst-icon.png",
                        "asset_class": "rwa",
                        "asset_subclass": "stablecoin",
                        "issuer_name": "Test Stablecoin Inc",
                        "uris": [
                            {
                                "uri": "example.org/ustst",
                                "category": "website",
                                "title": "USTST Official Website",
                            },
                            {
                                "uri": "example.org/ustst/reserves",
                                "category": "attestation",
                                "title": "Reserve Attestation Reports",
                            },
                            {
                                "uri": "example.org/ustst/docs",
                                "category": "docs",
                                "title": "USTST Documentation",
                            },
                        ],
                        "additional_info": {
                            "backing": "USD",
                            "reserve_ratio": "1:1",
                        },
                    }),
                ),
                client,
                mpt_issuer,
                autofill=True
            ),
            submit_and_wait(
                Batch(
                    account=domain_owner.address,
                    flags=BatchFlag.TF_ALL_OR_NOTHING,
                    raw_transactions=[
                        PermissionedDomainSet(
                            account=domain_owner.address,
                            accepted_credentials=[
                                Credential(
                                    issuer=domain_owner.address,
                                    credential_type=str_to_hex(cred_type)
                                )
                            ],
                        ),
                        CredentialCreate(
                            account=domain_owner.address,
                            subject=depositor.address,
                            credential_type=str_to_hex(cred_type),
                        ),
                    ],
                ),
                client,
                domain_owner,
                autofill=True
            )
        )

        mpt_issuance_id = mpt_create_result.result["meta"]["mpt_issuance_id"]

        # Get domain ID
        domain_owner_objects = await client.request(AccountObjects(
            account=domain_owner.address,
            ledger_index="validated",
        ))
        domain_id = next(
            node["index"]
            for node in domain_owner_objects.result["account_objects"]
            if node["LedgerEntryType"] == "PermissionedDomain"
        )

        # Step 2: Depositor accepts credential, authorizes MPT, and creates vault in parallel
        print("Setting up tutorial: 2/5", end="\r")

        _, vault_create_result = await asyncio.gather(
            submit_and_wait(
                Batch(
                    account=depositor.address,
                    flags=BatchFlag.TF_ALL_OR_NOTHING,
                    raw_transactions=[
                        CredentialAccept(
                            account=depositor.address,
                            issuer=domain_owner.address,
                            credential_type=str_to_hex(cred_type),
                        ),
                        MPTokenAuthorize(
                            account=depositor.address,
                            mptoken_issuance_id=mpt_issuance_id,
                        ),
                    ],
                ),
                client,
                depositor,
                autofill=True
            ),
            submit_and_wait(
                VaultCreate(
                    account=vault_owner.address,
                    asset={"mpt_issuance_id": mpt_issuance_id},
                    flags=VaultCreateFlag.TF_VAULT_PRIVATE,
                    domain_id=domain_id,
                    data=str_to_hex(json.dumps(
                        {"n": "LATAM Fund II", "w": "examplefund.com"}
                    )),
                    mptoken_metadata=encode_mptoken_metadata({
                        "ticker": "SHARE1",
                        "name": "Vault Shares",
                        "desc": "Proportional ownership shares of the vault",
                        "icon": "example.com/vault-shares-icon.png",
                        "asset_class": "defi",
                        "issuer_name": "Vault Owner",
                        "uris": [
                            {
                                "uri": "example.com/asset",
                                "category": "website",
                                "title": "Asset Website",
                            },
                            {
                                "uri": "example.com/docs",
                                "category": "docs",
                                "title": "Docs",
                            },
                        ],
                        "additional_info": {
                            "example_info": "test",
                        },
                    }),
                    assets_maximum="0",
                    withdrawal_policy=WithdrawalPolicy.VAULT_STRATEGY_FIRST_COME_FIRST_SERVE,
                ),
                client,
                vault_owner,
                autofill=True
            ),
        )

        # Extract vault_id and vault_share_mpt_issuance_id
        vault_node = next(
            node for node in vault_create_result.result["meta"]["AffectedNodes"]
            if "CreatedNode" in node and node["CreatedNode"].get("LedgerEntryType") == "Vault"
        )
        vault_id = vault_node["CreatedNode"]["LedgerIndex"]
        vault_share_mpt_issuance_id = vault_node["CreatedNode"]["NewFields"]["ShareMPTID"]

        # Step 3: Issuer sends payment to depositor
        print("Setting up tutorial: 3/5", end="\r")

        payment_result = await submit_and_wait(
            Payment(
                account=mpt_issuer.address,
                destination=depositor.address,
                amount={
                    "mpt_issuance_id": mpt_issuance_id,
                    "value": "10000",
                },
            ),
            client,
            mpt_issuer,
            autofill=True
        )

        if payment_result.result["meta"]["TransactionResult"] != "tesSUCCESS":
            print(f"\nPayment failed: {payment_result.result['meta']['TransactionResult']}", file=sys.stderr)
            sys.exit(1)

        # Step 4: Make an initial deposit so withdraw example has shares to work with
        print("Setting up tutorial: 4/5", end="\r")

        initial_deposit_result = await submit_and_wait(
            VaultDeposit(
                account=depositor.address,
                vault_id=vault_id,
                amount={
                    "mpt_issuance_id": mpt_issuance_id,
                    "value": "1000",
                },
            ),
            client,
            depositor,
            autofill=True
        )

        if initial_deposit_result.result["meta"]["TransactionResult"] != "tesSUCCESS":
            print(f"\nInitial deposit failed: {initial_deposit_result.result['meta']['TransactionResult']}", file=sys.stderr)
            sys.exit(1)

        # Step 5: Save setup data to file
        print("Setting up tutorial: 5/5", end="\r")

        setup_data = {
            "mpt_issuer": {
                "address": mpt_issuer.address,
                "seed": mpt_issuer.seed,
            },
            "mpt_issuance_id": mpt_issuance_id,
            "domain_owner": {
                "address": domain_owner.address,
                "seed": domain_owner.seed,
            },
            "domain_id": domain_id,
            "credential_type": cred_type,
            "depositor": {
                "address": depositor.address,
                "seed": depositor.seed,
            },
            "vault_owner": {
                "address": vault_owner.address,
                "seed": vault_owner.seed,
            },
            "vault_id": vault_id,
            "vault_share_mpt_issuance_id": vault_share_mpt_issuance_id,
        }

        with open("vault_setup.json", "w") as f:
            json.dump(setup_data, f, indent=2)

        print("Setting up tutorial: Complete!")


if __name__ == "__main__":
    asyncio.run(main())