Contract Interaction from Python

Full-Stack Starknet:

• [Part 1] 🚧 Getting Started in Cairo & Deploying with Nile
• [Part 2] 🐍 Contract Interaction with starknet.py (you are here)
• [Part 3] 👥 StarkNet Account Abstraction & Using Standard Contracts
• [Part 4] 💽 Local Devnet & Starknet.py's Account Capabilities
• [Part 5] 🎨 StarkNet Frontends w/ Cairopal & Argent X
• [Notes] 💰 Contract Costs & Why Our Design Needs Work

Completed code on GitHub here.

This part of the series will start the application that runs on our raspberry pi. By the end of this post, you'll know how to create and sign StarkNet transactions from python.

Setup

First, we'll initialize the project using Poetry:

sam@sam:~/fullstack-starknet/part2$ cd pi
sam@sam:~/fullstack-starknet/part2/pi$ poetry init

This should autogenerate a pyproject.toml. Here, we can add the dependencies and common dev tasks:

[tool.poetry.dependencies]
python = ">=3.7,<3.10"
fastapi = "^0.72.0"
"starknet.py" = "^0.1.5-alpha.0"
obd = "^0.7.1"
uvicorn = "^0.17.0"
mypy = "^0.931"

[tool.poetry.dev-dependencies]
black = "^21.12b0"
pytest = "^6.2.5"
taskipy = "^1.9.0"
safety = "^1.10.3"
isort = "^5.10.1"

[tool.taskipy.tasks]
audit = "poetry export --dev --format=requirements.txt --without-hashes | safety check --stdin --full-report"
dev = "uvicorn app.main:app --reload"
format = "isort app/ && black app/"
formatcheck = "black app/ --check"
test = "pytest"
typecheck = "mypy --config-file mypy.ini ."

Aside: FastAPI was chosen because it's quick to get started and gives us interactive API documentation out of the box for easy testing. This will all run locally, so the signing keys never leave the car.

Create an app folder with the following structure:

app
├── __init__.py
├── main.py

Themain.py module is the primary entrypoint to the application:

from fastapi import FastAPI

API_VERSION = "0.0.1"
app = FastAPI()


@app.get(path="/api")
def api_info():
    return {"version": API_VERSION}

Running the dev task will start the application locally, auto-reloading on save:

sam@sam:~/fullstack-starknet/part2/pi$ poetry run task dev
INFO:     Will watch for changes in these directories: ['/home/sam/fullstack-starknet/part2/pi']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [14934] using statreload
INFO:     Started server process [14936]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Now that we've got the application skeleton, we can start iterating on the functionality.

Intro to StarkNet.py

The starknet.py package is a clean way to access our contract from within our application logic. Although similar to the library used in our contract unit tests, starknet.py is a separate project with more fully-featured abstractions & tools.

Modifying our main.py a little bit, set up a temporary endpoint to try querying some testnet data:

from fastapi import FastAPI

from starknet_py.contract import Contract
from starknet_py.net.client import Client

API_VERSION = "0.0.1"
app = FastAPI()
starknet_client = Client("testnet")


@app.get(path="/api")
async def api_info():
    return {"version": API_VERSION}


@app.get(path="/api/tmp")
async def tmp_block():
    """A temporary function to demonstrate network interaction"""

    # Replace this with the block hash of your deployment from Part 1
    call_result = await starknet_client.get_block("0xf93145481a5ec656966de0ff6bfe507a2dec4fcbdb07a37cb8a2d3292305fb")
    return {"data": call_result}

This fetches the block containing our contract deployment (from Part 1):

sam@sam:~$ curl localhost:8000/api/tmp | jq
{
  "data": {
    "block_hash": "0xf4b05e3ec5bfc9e864b95bfcb2a8f1db04101c65635a92197d1002b6a52f04",
    "parent_block_hash": "0x1e9fa4b9cc5d32447826480e64daeb8465f2e3b080b2c8f39159760ce6576f1",
    "block_number": 46879,
    "state_root": "06343e6997bcb986380c0a831624889337b68bf77e838e64585a042b32955347",
    "status": "ACCEPTED_ON_L2",
    "transactions": [
      ...
      {
        "contract_address": "0x29af160331cb2c5898999034f51f3357243a36b93c7b696f7daf0711482458e",
        "contract_address_salt": "0x59d172b8d87ba423b70771d24b93f17afecd6787172ddffeb783aa977564d33",
        "constructor_calldata": [],
        "transaction_hash": "0x2fe91a7e15d1aa9890f7feabbae25864252a4a6fd0ef042e35250f71354073a",
        "type": "DEPLOY"
      }
    ],
    "timestamp": 1642473988,
    "transaction_receipts": [
      ...
      {
        "status": "ACCEPTED_ON_L2",
        "block_hash": "0xf4b05e3ec5bfc9e864b95bfcb2a8f1db04101c65635a92197d1002b6a52f04",
        "block_number": 46879,
        "transaction_index": 1,
        "transaction_hash": "0x2fe91a7e15d1aa9890f7feabbae25864252a4a6fd0ef042e35250f71354073a",
        "l2_to_l1_messages": [],
        "execution_resources": {
          "n_steps": 0,
          "builtin_instance_counter": {},
          "n_memory_holes": 0
        }
      }
    ]
  }
}

Calling Contract Functions

To start, we'll have two endpoints:

• POST /api/register – registers the host vehicle
• POST /api/commit – hashes recent vehicle data, then signs & commits it to StarkNet

Our primary application logic will live inside the POST /api/commit endpoint. Later, we'll configure a simple cron to hit it on a schedule.

Registration

For the sake of simplicity, we'll keep a lot of things hardcoded initially.

# Slightly updated imports
from fastapi import FastAPI, HTTPException
from starknet_py.contract import Contract
from starknet_py.net.client import Client

...

# TODO: read this from a config file / environment
contract_address="0x029af160331cb2c5898999034f51f3357243a36b93c7b696f7daf0711482458e"
vehicle_id = 1
private_key = 12345
public_key = 1628448741648245036800002906075225705100596136133912895015035902954123957052


@app.post(path="/api/register")
async def register():
    """Registers the car on chain, using configured data"""
    contract = await Contract.from_address(contract_address, starknet_client)

    # Calling a contract's function doesn't create a new transaction,
    # you get the function's result immediately. Use `call` for @views
    (owner,) = await contract.functions["get_owner"].call(vehicle_id)
    if owner != 0:
        raise HTTPException(status_code=403, detail="Vehicle already registered")

    # Writes (i.e. invokes) aren't accepted immediately.
    # Use `invoke` for @externals
    invocation = await contract.functions["register_vehicle"].invoke(
        vehicle_id=vehicle_id,
        owner_public_key=public_key,
        signer_public_key=public_key,
    )

    # ... but we can easily wait for it
    await invocation.wait_for_acceptance()

    return {"tx_hash": invocation.hash}

Calling the endpoint and waiting a little while, we get the transaction hash as a response:

sam@sam:~$ curl -X POST localhost:8000/api/register | jq
{
  "tx_hash": "0x5033ed47876fab5dd94521675a7705b5c5f9f74b87cc5751a0956dc46776167"
}

The explorer may be lagging behind a little, but it will eventually catch up and display the registration transaction. Returning to the #readContract tab of your contract's page, we can query for the owner of vehicle 1

… and see our public key as the owner!

Calling it again, we get the expected error:

sam@sam:~$ curl -X POST localhost:8000/api/register | jq
{
  "detail": "Vehicle already registered"
}

State Attestations

Moving on to the core loop, the POST /api/commit endpoint will be called periodically. This triggers:

• a reading of saved diagnostics data
• a hashing & signing of that data, along with the current vehicle nonce
• a transaction invoking the attest_state contract function

@app.post(path="/api/commit")
async def commit():
    """Commits a single state attestation on chain, using the most recent window of data."""
    contract = await Contract.from_address(contract_address, starknet_client)

    # Query current nonce to ensure we're signing the right payload
    (nonce,) = await contract.functions["get_nonce"].call(vehicle_id)

    # TODO: read from locally saved diagnostic data
    state_hash = 42424242
    
    # Prepare the function call without actually sending it yet.
    # This allows us to access felts exactly as they'll be serialized in calldata.
    prepared = contract.functions["attest_state"].prepare(
        vehicle_id=vehicle_id, nonce=nonce, state_hash=state_hash
    )

    # Hash and sign the transaction payload
    # Ordering: H( nonce, H( vehicle_id, H( state_hash, 0 ) ) )
    calldata = [
        # Each arg fits in a normal felt, so only the first element is filled.
        # Large numbers may span 2 or more elements here.
        prepared.arguments["state_hash"][0],
        prepared.arguments["vehicle_id"][0],
        prepared.arguments["nonce"][0],
    ]
    signature = sign_calldata(calldata, private_key)

    # Send it off & wait for confirmation
    invocation = await prepared.invoke(signature)
    await invocation.wait_for_acceptance()

    return {"nonce": nonce, "signature": signature, "tx_hash": invocation.hash}

After invoking it and waiting for a minute or two, we see:

sam@sam:~$ curl -X POST localhost:8000/api/commit | jq
{
  "nonce": 0,
  "signature": [
    1.0099925264838125e+75,
    1.9636227629845484e+75
  ],
  "tx_hash": "0x125d7092aca4aa03f8a5349312a0190e97d9917b2e9e474199cd42f4005fa6c"
}

Once that transaction confirms and is included in an L2 block, a query for the vehicle's nonce shows it's been updated, along with the state at nonce 0:

🎊 It works! We now have a deployed contract and the start of an application that can interact with it.

As for periodically reading from diagnotics from the OBD2 port, it's pretty straight forward and not very interesting sooo…

😅