tutorial

Building a NFT minting bot with Cardano, Telegram, and Python - part 3

Part 3: Using IPFS utils

ipfs_utils.py code overview

Outline

Intro

The first thing in our conversation with the bot will be to upload an image to IPFS, and then pin it. If you've read my previous post on this you will already know how this works.  I will go over the code here today, as it has a subtle but important change to my previous iterations.

You can find my previous posts here [both use the same version of the ipfs_utils.py]:

https://deafmice.com/blockfrost-io-script-example/
https://deafmice.com/mr-pinners-pinning-lab/

Code walk through

First you will need to add to your .env file the Blockfrost.io API key for your IPFS project.

# Blockfrost API
export BLOCKFROST_IPFS=YOUR_SECRET_KEY

Then add a quick line in the config.py.  You could just throw that line in the ipfs_utils.py file, but I like to keep a central location for changeable configurable variables.

# IPFS
BLOCKFROST_IPFS = os.getenv('BLOCKFROST_IPFS')

OK, now that we have the config taken care of let's begin with the actual code.

First, we import our modules. Luckily we just need the python module requests and logging.

# ipfs_utils.py

import config
import requests
import logging
logger = logging.getLogger(__name__)

The create_ipfs() function is where the change needs to happen.  On the Mr.Pinner web app, I was able to use a way to forward the image directly to the API, but for some reason, I was unable to do that with the Telegram API.  So you will see in the conversation part of the blog series where the bot actually has to download the image, then run this function, then remove the file download.

def create_ipfs(image):
    """ Uploads image to Blockfrost """
    with open(image, "rb") as file_upload:
        ipfs_create_url = "https://ipfs.blockfrost.io/api/v0/ipfs/add"
        files = {'file': file_upload}
        headers = {"project_id": f"{config.BLOCKFROST_IPFS}"}
        res = requests.post(ipfs_create_url,  files=files, headers=headers)
        res.raise_for_status()
        if res.status_code == 200:
            logging.info("Uploaded image to Blockfrost")
            logging.info(res.json())
            return res.json()
        else:
            logger.error("Something failed here? Upload to blockfrost failed")
            return False

The rest of the code is the same as we are just working with the API and an IPFS hash.


def pin_ipfs(ipfs_hash):
    """ Pins IPFS hash in Account """
    ipfs_pin_url = f"https://ipfs.blockfrost.io/api/v0/ipfs/pin/add/{ipfs_hash}"
    headers = {"project_id": f"{config.BLOCKFROST_IPFS}"}
    res = requests.post(ipfs_pin_url, headers=headers)
    res.raise_for_status()
    if res.status_code == 200:
        logger.info("Uploaded image to Blockfrost")
        logger.info(res.json())
        return res.json()
    else:
        logger.error("Something failed here? Upload to blockfrost failed")
        return False

I did not implement a remove or check section in the bot, but included the code as I may some time later.


def remove_ipfs(ipfs_hash):
    """ Removes pin from IPFS hash in Blockfrost """
    ipfs_remove_url = f"https://ipfs.blockfrost.io/api/v0/ipfs/pin/remove/{ipfs_hash}"
    headers = {"project_id": f"{config.BLOCKFROST_IPFS}"}
    res = requests.post(ipfs_remove_url, headers=headers)
    res.raise_for_status()
    if res.status_code == 200:
        logging.info("Removing pin worked")
        logging.info(res.json())
        return True
    else:
        logger.error("Something failed here? Remove Pin failed")
        return False


def check_ipfs(ipfs_hash):
    """ Check to see if the ipfs hash is accessible via
    ipfs.io and cloudflare """
    ipfs_gateways = [
        f"https://gateway.ipfs.io/ipfs/{ipfs_hash}",
        f"https://cloudflare-ipfs.com/ipfs/{ipfs_hash}"
    ]

    ipfs_status = []
    for gateway in ipfs_gateways:
        res = requests.get(gateway)
        logging.info(res.status_code)
        if res.status_code == 200:
            ipfs_status.append(True)
        else:
            ipfs_status.append(False)
    logging.info(f"Availability:  {ipfs_status}")
    return True

Wrap up

OK, we have some nifty IPFS functions we can have the bot work with.  

Next up, building the token_utils.py script.

Read next