Solutions
No Results

Create Publication

We are looking for publications that demonstrate building dApps or smart contracts!
See the full list of Gitcoin bounties that are eligible for rewards.

Solution

Getting started with Python Algorand SDK and FastAPI

Overview

Security warning
This project has not been tested and should never be used in production.

Requirements

You must have python 3 installed on your system. Also for this tutorial, you will need python3-venv. This package can be installed with the following command.

$ sudo apt-get install python3-venv

This manual assumes that Algorand Sandbox is already installed on your system.

Start work

First, create a root folder for the project.

cd ~
mkdir algorand-fastAPI
cd algorand-fastAPI

Then create and activate a new virtual environment.

python3 -m venv Algorand-fastAPI
./Algorand-fastAPI/bin/activate

Now you need to install FastAPI, Uvicorn, and py-algorand-sdk.

(Algorand-fastAPI) $ pip3 install fastapi
(Algorand-fastAPI) $ pip3 install uvicorn[standard]
(Algorand-fastAPI) $ pip3 install py-algorand-sdk

Create two new files: main.py and schemes.py

touch main.py 
touch schemes.py

In the file, main.py connect all the modules we need in the future, initialize the algod client, and the validator creates an instance of the application.

from typing import Optional
from fastapi import FastAPI

from algosdk import account, encoding , mnemonic 
from algosdk.future.transaction import PaymentTxn
from algosdk.future.transaction import AssetConfigTxn
from algosdk.error import WrongChecksumError ,WrongMnemonicLengthError
from algosdk.v2client import algod , indexer

from schemes import * 

algod_client=algod.AlgodClient("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","http://localhost:4001") #Initializing the algod client
indexer_client=indexer.IndexerClient("","http://localhost:8980") #initializing the validator

app = FastAPI() #Create a base application instance

Creating an account

Now we create a new endpoint of our API. This endpoint will be called by the http GET method.
In the body of our endpoint function, we call the account.generate_account() function, which returns a tuple of an account address and a private key. We convert this into a passphrase using the from_private_key method of the mnemonic class.

@app.get("/account")
def create_account():                                   #Create a function that creates an account
    private_key, address = account.generate_account()
    passphrase = mnemonic.from_private_key(private_key)

    return {"address":address,"passphrase":passphrase}

Now we can run this application with the command:

uvicorn main:app --reload

After downloading our application, the interactive documentation will be available to us at http://127.0.0.1:8000/docs#

Now when you send a get request, the server will create a new account and send the address along with the passphrase in the response

Get information about an account

To do this, create a new endpoint that will receive the address of the account and return information about it

@app.get("/account/{Address}")   
def get_account_info(Address:str):              #Create a function that returns information about an account by its ID
    info=algod_client.account_info(Address)

    return {"Address":info}

we can check that our endpoint works correctly with the interactive documentation

Transactions

Now we need to go back to the schemes.py file and create a schema for the transfer and asset creation transaction. The schema is needed so that the endpoint can check if the incoming data are of the expected type.

from pydantic import BaseModel

class Transaction(BaseModel):
    sender_address: str
    receiver_address: str
    passphrase: str
    amount: int
    note: str

class Asset(BaseModel):
    sender:str
    asset_name:str
    unit:str
    total:int
    default_frozen:bool
    decimals:int
    url:str
    manager:str
    reserve:str
    freeze:str
    clawback:str
    passphrase: str

Transfer of funds

Go back to the main.py file and create a new endpoint that receives data according to the Transaction schema.

@app.post("/transaction")
def create_transaction(transaction:Transaction):
    params = algod_client.suggested_params()

Now create an unsigned transaction and try to sign it.

unsigned_txn = PaymentTxn(transaction.sender_address, params, transaction.receiver_address, transaction.amount, None, transaction.note.encode()) #Create an unsigned transaction

    if transaction.amount < 100000:                      #check that the volume of transferred funds is greater than the minimum
        return {"amount":"amount less than the minimum"}
    try:
        signed_txn = unsigned_txn.sign(mnemonic.to_private_key(transaction.passphrase)) #trying to sign a transaction

    except WrongChecksumError:
        return {"passphrase":"Checksum error"}

    except ValueError:
        return {"passphrase":"unknown word in the passphrase"}

    except WrongMnemonicLengthError:
        return {"passphrase":"Incorrect size of the passphrase"}

When the transaction is signed we can send it to the blockchain through the algod client.

@app.post("/transaction")
def create_transaction(transaction:Transaction):
    params = algod_client.suggested_params()

Now create an unsigned transaction and try to sign it.

unsigned_txn = PaymentTxn(transaction.sender_address, params, transaction.receiver_address, transaction.amount, None, transaction.note.encode()) #Create an unsigned transaction

    if transaction.amount < 100000:                      #check that the number of transferred funds is greater than the minimum
        return {"amount":"amount less than the minimum"}
    try:
        signed_txn = unsigned_txn.sign(mnemonic.to_private_key(transaction.passphrase)) #trying to sign a transaction

    except WrongChecksumError:
        return {"passphrase":"Checksum error"}

    except ValueError:
        return {"passphrase":"unknown word in the passphrase"}

    except WrongMnemonicLengthError:
        return {"passphrase":"Incorrect size of the passphrase"}

When the transaction is signed we can send it to the blockchain.

    transaction_id = algod_client.send_transaction(signed_txn) #send the signed transaction to the network

    return transaction_id 

Next, let’s discuss how to create a new asset.

Asset creation

The process of creating an asset is not much different from the previous transaction, but has many more parameters:

  • sender (str) – address of the sender
  • sp (SuggestedParams) – suggested params from algod
  • index (int, optional) – index of the asset
  • total (int, optional) – total number of base units of this asset created
  • default_frozen (bool, optional) – whether slots for this asset in user accounts are frozen by default
  • unit_name (str, optional) – hint for the name of a unit of this asset
  • asset_name (str, optional) – hint for the name of the asset
  • manager (str, optional) – address allowed to change nonzero addresses for this asset
  • reserve (str, optional) – account whose holdings of this asset should be reported as “not minted”
  • freeze (str, optional) – account allowed to change frozen state of holdings of this asset
  • clawback (str, optional) – account allowed take units of this asset from any account
  • url (str, optional) – a URL where more information about the asset can be retrieved
  • strict_empty_address_check (bool, optional) – set this to False if you want to specify empty addresses. Otherwise, if this is left as True (the default), having empty addresses will raise an error, which will prevent accidentally removing admin access to assets or deleting the asset.
  • decimals (int, optional) – number of digits to use for display after decimal. If set to 0, the asset is not divisible. If set to 1, the base unit of the asset is in tenths. Must be between 0 and 19, inclusive. Defaults to 0.

This endpoint will have the following code. Note the similarities between both snippets.

@app.post("/asset")
def create_asset(asset:Asset):
    params = algod_client.suggested_params()
    unsigned_txn=AssetConfigTxn(sp=params,                                  #create an unsigned transaction to add a new asset
                                sender=asset.sender,
                                asset_name=asset.asset_name,
                                unit_name=asset.unit,
                                total=asset.total,
                                decimals=asset.decimals,
                                default_frozen=asset.default_frozen,
                                url=asset.url,
                                manager=asset.manager,
                                reserve=asset.reserve,
                                freeze=asset.freeze,
                                clawback=asset.clawback,
                                strict_empty_address_check=False)

    try:
        signed_txn = unsigned_txn.sign(mnemonic.to_private_key(asset.passphrase)) #trying to sign a transaction

    except WrongChecksumError:
        return {"passphrase":"Checksum error"}

    except ValueError:
        return {"passphrase":"unknown word in the passphrase"}

    transaction_id = algod_client.send_transaction(signed_txn) #send the signed transaction to the network
    return transaction_id

Endpoint to view all assets

@app.get("/assets")
def get_assets():                                   #Create a function that returns all existing assets
    return indexer_client.search_assets() 

Endpoint to view all transactions of the account

@app.get("/transactions/{account}")
def get_account_transactions(account:str):                         
    return indexer_client.search_transactions_by_address(account)

Endpoint to get information about a transaction by ID

@app.get("/transaction/{transaction_ID}")
def get_transaction_by_ID(transaction_ID:str):         
    return indexer_client.transaction(transaction_ID)

Data filtering

Now we can add the ability to filter transactions by timestamp and asset type.

The FastAPI framework allows you to add optional parameters to the endpoint request structure:
- add the parameters for the start and end timestamp in RFC 3339 format, and the asset ID

@app.get("/transactions/{account}")
def get_account_transactions(account:str, start_timestamp: Optional[str] = None , end_timestamp: Optional[str] = None , asset_id: Optional[int] = None):                          #Create a function that returns all transactions related to the account
    return indexer_client.search_transactions_by_address(account,start_time=start_timestamp,end_time=end_timestamp,asset_id=asset_id)

The endpoint is now able to issue transactions for an account for a certain period of time related to a specific asset.

Now let’s create an endpoint to get a list of transactions related to a particular asset, add optional parameters to filter by timestamp as shown in the previous code snippet.

@app.get("/assets/{asset_id}/transactions")
def get_assets(asset_id:int ,start_timestamp: Optional[str] = None , end_timestamp: Optional[str] = None):                                 
    return indexer_client.search_asset_transactions(asset_id , start_time = start_timestamp , end_time = end_timestamp) 

Obtaining a balance sheet for different assets

We start by creating an endpoint with two parameters, account address and asset ID. Next, we get information about the account by its address and check that the ID of the asset is not equal to 0. Otherwise, we return the balance of the account for the base asset in microAlgos.

Further, we try to find the asset we are looking for in the account. If we find the desired asset in an account, we retrieve the asset by its ID and form the response. If the asset is not found, we try to get the name of the asset. If this operation fails, we will inform the user that the asset does not exist. Otherwise, we form the answer with a zero balance in the sought-after asset.

@app.get("/account/{account}/balance")
def get_assets(Address:str,asset_id:int):                                  
    info=algod_client.account_info(Address)
    if asset_id==0:
        balance=algod_client.account_info(Address)["amount"]
        return {"balance":balance,"asset-id":0,"asset_name":"microAlgos"}

    for asset in info["assets"]:
        if asset["asset-id"]==asset_id:
            asset_name=indexer_client.search_assets(asset_id=asset_id)["assets"][0]["params"]["name"]

            return {"balance":asset["amount"],"asset-id":asset["asset-id"],"asset_name":asset_name}
    try:
        asset_name=indexer_client.search_assets(asset_id=asset_id)["assets"][0]["params"]["name"]
    except:
        return {"ERROR":"Asset not found"}
    return {"balance":0,"asset-id":asset_id,"asset_name":asset_name}

balance sheet for the underlying asset
This is the balance sheet for the underlying asset.

the balance for the asset we previously created
This is the balance for the asset we previously created.

That’s it! You can find the FastAPI docs here.

August 18, 2021