Building a Code-Sharing DApp with FastHTML

Introduction

FastHTML is a new Python framework for developing interactive web applications.

It's based on HTMX, and purportedly "scales down to a 6-line python file, and scales up to complex production apps".

I'm excited to see if I can use it to build full-stack decentralised applications with Algorand.

Code Highlighting

FastHTML makes it easy to integrate third-party JavaScript libraries, and comes with built-in support for highlight.js.

Reading the documentation gave me an idea for a code-sharing DApp, where users can store code snippets on the Algorand blockchain and share them publicly.

Each code snippet will be stored in a box, and rendered in the browser with syntax highlighting from highlight.js.

To start, let's display a simple code snippet on the home page:

from fasthtml.common import *

app = FastHTML(
    hdrs=(
        picolink,
        HighlightJS(langs=["python", "javascript"]),
    ),
)

@app.route("/")
def get():
    title = "Code Snippet"
    code_snippet = Div(Pre(Code('print("Hello, World!")')), id="code-snippet")
    return (
        Title(title),
        Main(
            H1(title),
            code_snippet,
            cls="container",
        ),
    )

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        port=5000,
        reload=True,
    )

This renders:

Perfect.

But we want to store the code snippets on Algorand, so we'll need to write a basic contract.

The Smart Contract

The smart contract will be written in Algorand Python.

To keep things simple, we'll add a single method that allows any user to store data in the application's box storage.

The key for each box will be the hash of its contents (the code snippet).

Content addressing through hashes has become a widely-used means of connecting data in distributed systems.

Source: https://ipld.io/

Storage costs will be covered on a pay-as-you-go basis.

When a box with name n and size s is created, the MBR is raised by 2500 + 400 * (len(n)+s) microAlgos. When the box is destroyed, the minimum balance requirement is decremented by the same amount.

Source: https://developer.algorand.org/articles/smart-contract-storage-boxes/

I assume these fees could change at some point in the future, so rather than calculating the balance with the formula above, the contract will simply check that the payment amount received is >= the difference between its starting and ending minimum balance requirement:

from algopy import ARC4Contract, String, arc4, BoxMap, op, Bytes, Global, gtxn
from typing import TypeAlias, Literal

Hash: TypeAlias = Bytes


class Storage(ARC4Contract):
    def __init__(self) -> None:
        self.content = BoxMap(Hash, String)

    @arc4.abimethod()
    def store(self, payment: gtxn.PaymentTransaction, code: String) -> None:
        mbr_before = Global.current_application_address.min_balance

        content_hash = op.sha256(code.bytes)
        # Saves the user paying txn fee if the content is already stored
        assert content_hash not in self.content, "Content already stored"
        self.content[content_hash] = code

        mbr_after = Global.current_application_address.min_balance

        assert (
            payment.receiver == Global.current_application_address
        ), "Receiver must be the app address"
        assert (
            payment.amount >= mbr_after - mbr_before
        ), "Payment amount must >= the minimum balance requirement delta"

There are no fees for failed transactions on Algorand, so we may as well throw an error if a user tries to store a snippet that already exists.

Deploying the Contract

from fastblock.blockchain.projects.blockchain.smart_contracts.artifacts.storage.client import (
    StorageClient,
)

from base64 import b64decode
from algokit_utils import (
    Account,
    TransferParameters,
    transfer,
)
from algosdk.transaction import PaymentTxn
from algosdk.v2client.algod import AlgodClient
from algosdk.v2client.indexer import IndexerClient
from hashlib import sha256

from algokit_utils import OnSchemaBreak, OnUpdate


def deploy_idempotent(
    account: Account, algod: AlgodClient, indexer: IndexerClient
) -> StorageClient:
    """Deploys the storage contract if it hasn't already been deployed.

    Funds the application account with the minimum balance required.

    Args:
        account (Account): The account that will deploy the contract.
        algod (AlgodClient): The Algod client.
        indexer (IndexerClient): The Indexer client.

    Returns:
        StorageClient: The storage contract client.
    """
    storage = StorageClient(
        algod,
        creator=account,
        indexer_client=indexer,
    )
    storage.deploy(
        on_schema_break=OnSchemaBreak.AppendApp,
        on_update=OnUpdate.AppendApp,
    )

    if (current_balance := balance(algod, storage.app_address)) < 100_000:
        print(
            f"Funding storage app account with {(delta := 100_000 - current_balance)} microAlgos"
        )
        transfer(
            algod,
            TransferParameters(
                from_account=account,
                to_address=storage.app_address,
                micro_algos=delta,
            ),
        )

    return storage

Ignore the horrendously long import - I'm too lazy to simplify the generated folder structure 🙃

The app deployment is handled by algokit-utils, and is idempotent by default.

The latter part of the function transfers enough funds to cover the minimum balance requirement for the app account (100,000 MicroAlgos).

Interacting With the Contract

I've quickly defined a few additional functions to help read data from the blockchain:

def balance(algod: AlgodClient, account: Account | str) -> int:
    """Returns the balance of the specified account.

    Args:
        algod (AlgodClient): The Algod client.
        account (Account | str): The account or address.

    Returns:
        int: The account balance in MicroAlgos.
    """
    return algod.account_info(getattr(account, "address", account))["amount"]


def box_name(code: str) -> bytes:
    """Hashes the code to derive the box name (key).

    Args:
        code (str): The code to hash.

    Returns:
        bytes: The box name (key).
    """
    return b"content" + sha256(code.encode()).digest()

def storage_cost(code: str) -> int:
    """Calculates the box storage cost for the code provided.

    Reference: https://developer.algorand.org/articles/smart-contract-storage-boxes/
    Formula: 2500 + 400 * (len(n)+s)

    Args:
        code (str): The code to calculate the storage cost for.

    Returns:
        int: The storage cost in MicroAlgos.
    """
    name_length = 39 # 7 byte "content" prefix + 32 byte hash
    return 2_500 + 400 * (name_length + len(code))

def storage_payment_txn(
    storage: StorageClient, account: Account | str, algod: AlgodClient, amount: int
) -> PaymentTxn:
    """Returns a payment transaction object to fund the storage contract.

    Args:
        storage (StorageClient): The storage contract client.
        account (Account | str): The account to pay from.
        algod (AlgodClient): The Algod client.
        amount (int): The amount to pay in MicroAlgos.

    Returns:
        PaymentTxn: The payment transaction object.
    """
    return PaymentTxn(
        sender=getattr(account, "address", account),
        sp=algod.suggested_params(),
        receiver=storage.app_address,
        amt=amount,
    )


@lru_cache(maxsize=32)
def fetch_box(app_id: int, indexer: IndexerClient, box_name: bytes) -> str:
    """Returns the value of the specified box.

    Args:
        app_id (int): The storage contract app ID.
        indexer (IndexerClient): The Indexer client.
        box_name (bytes): The box name (key).

    Returns:
        str: The box value.
    """
    return b64decode(
        indexer.application_box_by_name(application_id=app_id, box_name=box_name)[
            "value"
        ]
    ).decode()


def fetch_boxes(app_id: int, indexer: IndexerClient) -> list[str]:
    """Returns the values of all boxes in the storage contract,
    up to the default limit.

    Args:
        app_id (int): The storage contract app ID.
        indexer (IndexerClient): The Indexer client.

    Returns:
        list[str]: The box values.
    """
    return [
        fetch_box(app_id, indexer, b64decode(x["name"]))
        for x in indexer.application_boxes(application_id=app_id)["boxes"]
    ]

Hopefully the function definitions are descriptive enough, so I'll move on!

Piecing It Together

Back in main.py, let's use the functions discussed previously to deploy the contract to a LocalNet instance:

def dependencies() -> tuple[AlgodClient, IndexerClient, StorageClient]:
    load_dotenv(".env.localnet")

    algod = get_algod_client()
    indexer = get_indexer_client()
    account = get_localnet_default_account(algod)

    storage = deploy_idempotent(account, algod, indexer)

    print(f"{storage.app_id = }")

    code = dedent(
        """\
        print("Welcome to Code Snippet!")
        '''
        This code is stored on the Algorand blockchain :)
        Connect your wallet to upload and share your own snippets!
        '''
        """
    )

    ptxn = storage_payment_txn(storage, account, algod, storage_cost(code))

    try:
        storage.store(
            payment=TransactionWithSigner(
                ptxn, AccountTransactionSigner(account.private_key)
            ),
            code=code,
            transaction_parameters=TransactionParameters(boxes=[(0, box_name(code))]),
        )
    except LogicError as e:
        # Might be a better way to do this
        if e.pc == 88:
            print("Content already stored")
        else:
            raise e

    return algod, indexer, storage

Here we call the store method to upload the first snippet, and make a payment to the contract to cover the storage costs.

If a logic error is thrown with program counter 88, it means that snippet has already been uploaded.

We can call this function in the home page section:

@app.route("/")
def get():
    title = "Code Snippet"
    algod, indexer, storage, account = dependencies()

And then read the snippet from box storage:

@app.route("/")
def get():
    title = "Code Snippet"
    algod, indexer, storage, account = dependencies()
    default_code = next(
        iter(fetch_boxes(app_id=storage.app_id, indexer=indexer)),
        "# No snippets created yet :(",
    )
    code_snippet = Div(Pre(Code(default_code)), id="code-snippet")
    return (
        Title(title),
        Main(
            H1(title),
            code_snippet,
            cls="container",
        ),
    )

Which renders:

Good progress!

We now have a fully functioning app that stores code on the blockchain, and renders it in the browser.

But for this to be useful, we need a way for users to connect their wallet and upload their own snippets.

Connecting a Wallet

To start, let's create a wallet connect button on the home page:

@app.route("/")
def get():
    title = "Code Snippet"
    algod, indexer, storage, account = dependencies()
    default_code = next(
        iter(fetch_boxes(app_id=storage.app_id, indexer=indexer)),
        "# No snippets created yet :(",
    )
    code_snippet = Div(Pre(Code(default_code)), id="code-snippet")
    return (
        Title(title),
        Header(
            Div(
                Button(
                    "Connect", id="connect-button", cls="contrast", style="width:125px; margin-bottom: 10px;"
                ),
                cls="container",
                style="display: flex; justify-content: flex-end; border-bottom: 1px solid #e5e7eb;",
            ),
        ),
        Main(
            H1(title),
            code_snippet,
            cls="container",
        ),
    )

To keep the bundle size small, I've chosen to only integrate with Pera wallet (the most popular wallet provider).

The code is copied from the official vanilla JS example, with a few small changes.

I used webpack to bundle the JavaScript, and then loaded it in the FastHTML app as follows:

@app.route("/")
def get():
    title = "Code Snippet"
    algod, indexer, storage, account = dependencies()
    default_code = next(
        iter(fetch_boxes(app_id=storage.app_id, indexer=indexer)),
        "# No snippets created yet :(",
    )
    code_snippet = Div(Pre(Code(default_code)), id="code-snippet")
    pera = Script(
        open(Path(__file__).parents / "bundle.js").read(),
        type="module",
    )
    return (
        Title(title),
        Header(
            Div(
                Button(
                    "Connect", id="connect-button", cls="contrast", style="width:125px; margin-bottom: 10px;"
                ),
                cls="container",
                style="display: flex; justify-content: flex-end; border-bottom: 1px solid #e5e7eb;",
            ),
        ),
        Main(
            H1(title),
            code_snippet,
            cls="container",
        ),
        pera,
    )

Clicking the 'connect' button now triggers the Pera wallet modal:

Accepting Input

Once a user has connected their wallet, they should be able to upload a code snippet.

Let's create an input element that's visible when a wallet is connected, and hidden otherwise:

upload_section = Div(
    Textarea(
        id="code-input",
        name="code",
        placeholder="Paste your code here...",
        style="min-height: 100px;",
    ),
    Button("Upload", id="upload-button", cls="contrast"),
    id="upload-section",
    style="display: none;",
)

Main(
    H1(title),
    code_snippet,
    upload_section,
    cls="container",
),

When visible, the upload section looks like this:

To toggle the visibility, we can add the following code to the JavaScript file that contains the Pera wallet connect logic:

const uploadSection = document.getElementById("upload-section");


function handleConnectWalletClick(event) {
  ...
  uploadSection.style.display = "block";
  ...
}

function handleDisconnectWalletClick(event) {
  ...
  uploadSection.style.display = "none";
  ...
}

The end product looks like this:

Signing a Transaction

To upload a snippet, a user needs to sign two transactions in a group: one to make a payment to the storage contract, and one to call the application.

The value from the upload section’s input element needs to be passed to the application call transaction as an argument.

The sensible thing to do here would probably be to use AlgoKit to generate a TypeScript client, and then bundle it for the browser.

But that’s no fun. Let's create an endpoint that returns the encoded transactions, ready to sign:

@app.get("/txns")
def get_txns(sender: str, code: str):
    """Returns the encoded transactions to sign.

    Args:
        sender (str): The sender's address.
        code (str): The code snippet to upload.

    Returns:
        str: A JSON array of base64-encoded transactions.
    """
    algod, indexer, storage = dependencies()
    storage.sender = sender

    unsigned_txns = (
        storage.compose()
        .store(
            payment=TransactionWithSigner(
                storage_payment_txn(storage, sender, algod, storage_cost(code)),
                AccountTransactionSigner(""),
            ),
            code=code,
            transaction_parameters=TransactionParameters(boxes=[(0, box_name(code))]),
        )
        .build()
        .build_group()
    )
    return json.dumps([msgpack_encode(x.txn) for x in unsigned_txns])

Then back in the JavaScript file we can add a listener to the upload button:

uploadButton.addEventListener("click", (event) => {
  event.preventDefault();

  // Set loading spinner
  uploadButton.setAttribute("aria-busy", "true");

  console.log("Requesting signature from account", accountAddress);

  if (codeInput.value) {
    const contentHash = CryptoJS.SHA256(codeInput.value).toString();
    fetch(
      `/txns?sender=${accountAddress}&code=${encodeURIComponent(
        codeInput.value
      )}`,
      {
        headers: {
          "Content-Type": "application/json",
        },
      }
    )
      .then((response) => response.json())
      .then(async (data) => {
        const txns = data.map((item) => {
          return {
            txn: algosdk.decodeUnsignedTransaction(Buffer.from(item, "base64")),
            signers: [accountAddress],
          };
        });

        try {
          const signedTxns = await peraWallet.signTransaction([txns]);
          const { txId } = await algod.sendRawTransaction(signedTxns).do();
          console.log("Transaction ID:", txId);
          console.log("Content hash:", contentHash);

          // Remove loading spinner
          uploadButton.setAttribute("aria-busy", "false");

          if (txId && contentHash) {
            const currentUrl = new URL(window.location.href);
            currentUrl.searchParams.set("id", contentHash);
            currentUrl.searchParams.set("uploaded", true);
            // Redirect
            window.location.href = currentUrl.toString();
          }
        } catch (error) {
          console.log("Transaction signing failed:", error);
        }
      })
      .catch((error) => console.error("Error:", error));
  }
});

This sends a request to the /txns endpoint and passes the response to Pera wallet to sign.

Nearly there!

Viewing Snippets

Once a user has uploaded a code snippet, they need to be able to view it for confirmation and share it with others.

We can achieve this by using an optional query parameter on the index route:

@app.route("/")
def get(id: str | None = None):
    ...

If an ID (hash) is provided, we try to fetch it from box storage and render it:

@app.route("/")
def get(id: str | None = None):
    ...
    code = next(
        iter(fetch_boxes(app_id=storage.app_id, indexer=indexer)),
        "# No snippets created yet :(",
    )
    if id:
        try:
            code = fetch_box(
                app_id=storage.app_id,
                indexer=indexer,
                box_name=b"content" + bytes.fromhex(id),
            )
        except IndexerHTTPError:
            print(f"Invalid box name: {id}")

    code_snippet = Div(Pre(Code(code)), id="code-snippet")

If that fails or an ID is not provided, we render the default snippet.

But there’s one complication. Boxes are fetched using the indexer, and the indexer takes a bit of time to update (usually 5-10 seconds).

So when a user uploads a new snippet, we need to show a temporary loading screen and replace it when the box value is available.

Polling

Let’s create a new function to handle the various scenarios for fetching a snippet:

def snippet(id: str | None, uploaded: bool, request_count: int = 0):
    """Reads a snippet from box storage.

    If no ID is provided, the default snippet is displayed.
    Otherwise, the indexer is polled every second until the snippet is found.
    If the snippet is not found after 10 requests, an error message is displayed.

    Args:
        id (str | None): The snippet ID to search for.
        uploaded (bool): True if the snippet has been uploaded by the user.
        request_count (int, optional): The number of requests made. Defaults to 0.
    """
    algod, indexer, storage = dependencies(app_id=2302451247)

    id = (
        id or "10304786245d1a14092de73d13e72f83495122882122df482ce5bcde3a063a82"
    )  # Default snippet

    if request_count > 10:
        return Div(Card(f"Could not find snippet {id}."), id="code-snippet")
    try:
        code = fetch_box(
            app_id=storage.app_id,
            indexer=indexer,
            box_name=b"content" + bytes.fromhex(id),
        )
        return Div(
            P(
                "Congratulations! 🥳 Your snippet is stored on Algorand:"
                if uploaded
                else ""
            ),
            Pre(Code(code)),
            id="code-snippet",
        )
    except IndexerHTTPError:
        return Div(
            Card(
                P("Waiting for indexer...", aria_busy="true"),
                Progress(value=request_count, max=10),
            ),
            id="code-snippet",
            hx_post=f"/snippet/{id}/{uploaded}/{request_count + 1}",
            hx_trigger="every 1s",
            hx_swap="outerHTML",
        )

First, if no snippet ID is provided, the function will return the default snippet.

If an ID is provided, then we try to fetch the snippet from box storage.

If it’s not available in the indexer yet, we’ll show a progress bar:

And finally, if the snippet still isn’t available in the indexer after 10 seconds, we display the error message seen in the video above.

The magic here is the mutual recursion from this code in snippet:

hx_post=f"/snippet/{id}/{uploaded}/{request_count + 1}",
hx_trigger="every 1s",
hx_swap="outerHTML",

Which calls the snippet endpoint every second.

That endpoint in turn calls the snippet function:

@app.post("/snippet/{id}/{uploaded}/{request_count}")
def post(id: str, uploaded: bool, request_count: int):
    return snippet(id, uploaded, request_count)

And so it goes on until the recursive base case is satisfied:

if request_count > 10:
    return Div(Card(f"Could not find snippet {id}."), id="code-snippet")

Piecing It All Together

The final user flow looks like this:

I deployed the app to Vercel because I already have an account there.

You can check it out here: https://fastblock.vercel.app/

And see the source code here: https://github.com/code-alexander/fastblock/

Happy coding! 🥳