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}
This is the balance sheet for the underlying asset.
This is the balance for the asset we previously created.
That’s it! You can find the FastAPI docs here.