Overview

The Python v3.11 Runtime enables you to build and deploy serverless functions using the Python v3.11 environment. This runtime provides access to popular packages and web3 libraries, as well as the Python core modules. You can also include additional packages by uploading a zip file containing your function code and dependencies.

Available Packages

The Python v3.11 runtime includes the following useful packages and web3 libraries which can be used inside your function:

• web3==6.20.1
• eth_utils==4.1.1
• pandas==2.2.2
• numpy==2.0.1
• requests==2.32.3
• cryptography==43.0.0
• aiohttp==3.10.1
• python-dotenv==1.0.1
• ujson==5.10.0

Available Packages

The Python v3.11 runtime includes the following standard library modules:

• abc
• argparse
• array
• asyncio
• base64
• binascii
• bisect
• calendar
• cmath
• collections
• concurrent
• contextlib
• copy
• csv
• dataclasses
• datetime
• decimal
• difflib
• dis
• enum
• errno
• faulthandler
• filecmp
• fnmatch
• fractions
• functools
• gc
• getopt
• getpass
• gettext
• glob
• gzip
• hashlib
• heapq
• hmac
• html
• http
• imaplib
• imghdr
• importlib
• inspect
• io
• ipaddress
• itertools
• json
• keyword
• linecache
• locale
• logging
• lzma
• math
• multiprocessing
• netrc
• numbers
• operator
• os
• pathlib
• pickle
• pkgutil
• platform
• plistlib
• pprint
• profile
• pstats
• pyclbr
• pydoc
• queue
• random
• re
• readline
• reprlib
• sched
• secrets
• selectors
• shelve
• shlex
• shutil
• smtplib
• sndhdr
• socket
• sqlite3
• ssl
• stat
• statistics
• string
• subprocess
• sys
• tempfile
• textwrap
• threading
• time
• timeit
• trace
• traceback
• typing
• unittest
• urllib
• uuid
• venv
• warnings
• weakref
• webbrowser
• xml
• zipfile
• zipimport
• zlib

For more details, you can refer to the Python GitHub repository.

Using the Code Editor

Within the code editor, you can write and test your function using the standard library modules listed above. Here is a simple example of a Python function:

def main(params):
    # Extract dataset and network from metadata in params
    dataset = params['metadata']['dataset']
    network = params['metadata']['network']

    return {
        'message': f'This is data from the {dataset} dataset on the {network} network.',
        'params': params
    }

Uploading a Zip File

If you need to include additional pip packages, you can upload your function as a zip file. Follow these steps:

• Step 1: Create your function code in __main__.py.
• Step 2: Create a requirements.txt file with your dependencies.
• Step 3: Run pip install -r requirements.txt -t . to install the dependencies into your project directory.
• Step 4: Zip the contents of your project directory, including __main__.py, requirements.txt, and the installed packages. Note: Make sure that you do not zip the containing folder. Instead, select the files you want to include in your zip, for example the __main__.py, .env, and requirements.txt files as well as the installed packages, and compress them into an archive.
• Step 5: Upload the zip file in the function editor.
• Step 6: Test this function with the Block with Receipts dataset.

Example requirements.txt:

requests
web3
python-dotenv

Example __main__.py:

import os
import requests
from dotenv import load_dotenv
from web3 import Web3

# Load environment variables
load_dotenv()
QUICKNODE_URL = os.getenv('QUICKNODE_URL')

# Initialize Web3
web3 = Web3(Web3.HTTPProvider(QUICKNODE_URL))

# Sample ABI for common token functions
SAMPLE_TOKEN_ABI = [
    {
        "constant": True,
        "inputs": [],
        "name": "symbol",
        "outputs": [{"name": "", "type": "string"}],
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [{"name": "_tokenId", "type": "uint256"}],
        "name": "tokenURI",
        "outputs": [{"name": "", "type": "string"}],
        "type": "function"
    }
]

def call_contract(contract_address, function_signature, *args):
    checksum_address = Web3.to_checksum_address(contract_address)
    contract = web3.eth.contract(address=checksum_address, abi=SAMPLE_TOKEN_ABI)
    function = getattr(contract.functions, function_signature)
    return function(*args).call()

def get_token_symbol(contract_address):
    try:
        return call_contract(contract_address, 'symbol')
    except Exception as e:
        print(f"Error getting symbol: {str(e)}")
        return "Unknown"

def get_token_uri(contract_address, token_id):
    try:
        return call_contract(contract_address, 'tokenURI', token_id)
    except Exception as e:
        print(f"Error getting tokenURI: {str(e)}")
        return "Unknown"

def process_block_with_receipts(data):
    results = []
    for entry in data:
        block = entry.get('block', {})
        receipts = entry.get('receipts', [])
        
        block_summary = {
            'blockNumber': block.get('number'),
            'transactionCount': len(receipts),
            'logEventCount': sum(len(receipt.get('logs', [])) for receipt in receipts),
            'tokenTransfers': []
        }
        
        transfer_event_signature = web3.keccak(text="Transfer(address,address,uint256)").hex()
        
        for receipt in receipts:
            for log in receipt.get('logs', []):
                if log['topics'][0] == transfer_event_signature:
                    from_address = Web3.to_checksum_address(log['topics'][1][-40:])
                    to_address = Web3.to_checksum_address(log['topics'][2][-40:])
                    contract_address = Web3.to_checksum_address(log['address'])
                    if len(log['topics']) == 3:
                        # ERC20 Transfer
                        value = int(log['data'], 16)
                        symbol = get_token_symbol(contract_address)
                        block_summary['tokenTransfers'].append({
                            'type': 'ERC20',
                            'from': from_address,
                            'to': to_address,
                            'value': str(value),
                            'symbol': symbol
                        })
                    elif len(log['topics']) == 4:
                        # ERC721 Transfer
                        token_id = int(log['topics'][3], 16)
                        token_uri = get_token_uri(contract_address, token_id)
                        block_summary['tokenTransfers'].append({
                            'type': 'ERC721',
                            'from': from_address,
                            'to': to_address,
                            'tokenId': token_id,
                            'tokenURI': token_uri
                        })
        results.append(block_summary)
    
    return {'results': results}

def main(params):
    data = params.get('data', [])
    return process_block_with_receipts(data)

def handler(params):
    return main(params)

Example .env:

# QUICKNODE_URL=https://your-quicknode-url. Get one here: https://dashboard.quicknode.com/endpoints
QUICKNODE_URL="XXX"