Fix lint issues.

pull/659/head
Andrey 2022-09-07 14:48:33 +03:00
rodzic 8855097728
commit 7d073c3e63
2 zmienionych plików z 16 dodań i 14 usunięć

Wyświetl plik

@ -3,7 +3,8 @@ import json
import hashlib import hashlib
import itertools import itertools
import logging import logging
from typing import List, Any, Optional from this import d
from typing import Dict, List, Any, Optional
from uuid import UUID from uuid import UUID
from moonstreamdb.blockchain import AvailableBlockchainType from moonstreamdb.blockchain import AvailableBlockchainType
@ -173,20 +174,22 @@ def parse_jobs(
access_id: UUID, access_id: UUID,
): ):
contracts_ABIs = {} contracts_ABIs: Dict[str, Any] = {}
contracts_methods = {} contracts_methods: Dict[str, Any] = {}
calls = {0: []} calls: Dict[int, Any] = {0: []}
web3_client = _retry_connect_web3( web3_client = _retry_connect_web3(
blockchain_type=blockchain_type, access_id=access_id blockchain_type=blockchain_type, access_id=access_id
) )
if block_number is None: if block_number is None:
block_number = web3_client.eth.get_block("latest").number block_number = web3_client.eth.get_block("latest").number # type: ignore
block_timestamp = web3_client.eth.get_block(block_number).timestamp block_timestamp = web3_client.eth.get_block(block_number).timestamp # type: ignore
multicaller = Multicall2(web3=web3_client, contract_address=Multicall2_address) multicaller = Multicall2(
web3_client, web3_client.toChecksumAddress(Multicall2_address)
)
multicall_method = multicaller.tryAggregate multicall_method = multicaller.tryAggregate
@ -260,10 +263,10 @@ def parse_jobs(
# generate interface # generate interface
interfaces[contract_address] = web3_client.eth.contract( interfaces[contract_address] = web3_client.eth.contract(
address=contract_address, abi=abis address=web3_client.toChecksumAddress(contract_address), abi=abis
) )
responces = {} responces: Dict[str, Any] = {}
# # create chunks of calls # # create chunks of calls

Wyświetl plik

@ -2,14 +2,13 @@ import getpass
import os import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union from typing import Any, Callable, Dict, List, Optional, Tuple, Union
import web3
from eth_account.account import Account # type: ignore from eth_account.account import Account # type: ignore
from eth_abi import encode_single, decode_single from eth_abi import encode_single, decode_single
from eth_typing.evm import ChecksumAddress from eth_typing.evm import ChecksumAddress
from eth_utils import function_signature_to_4byte_selector from eth_utils import function_signature_to_4byte_selector
from hexbytes.main import HexBytes from hexbytes.main import HexBytes
from web3 import Web3 from web3 import Web3
from web3.contract import Contract, ContractFunction from web3.contract import ContractFunction
from web3.providers.ipc import IPCProvider from web3.providers.ipc import IPCProvider
from web3.providers.rpc import HTTPProvider from web3.providers.rpc import HTTPProvider
from web3.types import ABI, Nonce, TxParams, TxReceipt, Wei from web3.types import ABI, Nonce, TxParams, TxReceipt, Wei
@ -118,7 +117,7 @@ def deploy_contract_from_constructor_function(
contract_abi: List[Dict[str, Any]], contract_abi: List[Dict[str, Any]],
deployer: ChecksumAddress, deployer: ChecksumAddress,
deployer_private_key: str, deployer_private_key: str,
constructor: ContractConstructor, constructor: ContractFunction,
) -> Tuple[HexBytes, ChecksumAddress]: ) -> Tuple[HexBytes, ChecksumAddress]:
""" """
Deploys smart contract to blockchain from constructor ContractFunction Deploys smart contract to blockchain from constructor ContractFunction
@ -205,7 +204,7 @@ def cast_to_python_type(evm_type: str) -> Callable:
class FunctionInput: class FunctionInput:
def __init__(self, name: str, value: any, solidity_type: str): def __init__(self, name: str, value: Any, solidity_type: str):
self.name = name self.name = name
self.value = value self.value = value
self.solidity_type = solidity_type self.solidity_type = solidity_type
@ -236,7 +235,7 @@ class FunctionSignature:
self.fourbyte = function_signature_to_4byte_selector(self.signature) self.fourbyte = function_signature_to_4byte_selector(self.signature)
def encode_data(self, args=None) -> str: def encode_data(self, args=None) -> bytes:
return ( return (
self.fourbyte + encode_single(self.input_types_signature, args) self.fourbyte + encode_single(self.input_types_signature, args)
if args if args