Home

Introduction

The Amazon Managed Blockchain (AMB) is a blockchain service that makes it easy to create and manage scalable blockchain networks using open-source frameworks like Hyperledger Fabric and Ethereum. With AMB, you can create your own blockchain network with just a few clicks, allowing for faster deployment of blockchain solutions for various use cases.

Additionally, AMB provides a range of features including automatic scaling, built-in security, and a fully managed service. By using AMB, you can focus on building your blockchain applications and not worry about the underlying infrastructure.

from web3.types import Middleware, RPCEndpoint, RPCResponse
from web3.providers.rpc import HTTPProvider
from requests_auth_aws_sigv4 import AWSSigV4
from eth_typing import URI
from settings import settings
from web3 import Web3
from typing import Any
import requests

def __init__(self, endpoint_uri: URI) -> None:
    self.aws_auth = AWSSigV4(
        'managedblockchain',
        aws_access_key_id=settings.aws_access_key_id,
        aws_secret_access_key=settings.aws_secret_access_key,
        region=settings.aws_region # us-east-1
    )
    self.session = requests.Session()
    super().__init__(endpoint_uri)

This is the code for the “init” part of a class. The “init” section is responsible for setting up the foundation of the class. Here, we create the “aws_auth” (credentials) and session. The “aws_auth” section is used to authenticate with AWS and the session is used to manage the connection to AWS.

def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
        request_data = self.encode_rpc_request(method, params).decode()
        raw_response = self.make_custom_post_request(
            self.endpoint_uri,
            request_data,
            **self.get_request_kwargs()
        )
        response = self.decode_rpc_response(raw_response)
        return response
    
def get_request_kwargs(self) -> dict:
    return {
        "headers": {
            "Content-Type": "application/json",
            "X-Amz-Target": "ManagedBlockchain_v2018_09_24.CreateMember",
        }
    }

def make_custom_post_request(self,
        endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any) -> bytes:
    kwargs.setdefault('timeout', 10)
    response = self.session.post(endpoint_uri, data=data,
                            *args, **kwargs, auth=self.aws_auth) 
    response.raise_for_status()
    return response.content

AMBHTTPProvider inherits HTTPProvider and make_request operates within HTTPProvider. Therefore, commands invoked using AMBHTTPProvider are automatically passed to make_request within HTTPProvider. By default, make_request calls make_post_request, but in this code, it is configured to use make_custom_post_request for awa_auth authentication.

Usage

if __name__ == "__main__":
    provider = AMBHTTPProvider(settings.endpoint_url)
    w3 = Web3(provider)
    connection_valid = w3.is_connected()
    print(connection_valid)
    print(w3.eth.block_number)

As you can see from the code above, the web3 library can be used with web3py by connecting it to AMBHTTPProvider. This allows for easy interaction with Ethereum-based networks, enabling developers to create decentralized applications that are secure, transparent, and scalable.