Core utilities¶

Shared constants, filters, decorators, and helpers used across the SDK.

Constants¶

ARTIFACTS_FOLDER `module-attribute` ¶

ARTIFACTS_FOLDER = join(
    dirname(dirname(__file__)), "artifacts"
)

Path to the directory containing compiled smart contract artifacts (ABIs and bytecode).

DEFAULT_CONFIRMATION_POLL_INTERVAL `module-attribute` ¶

DEFAULT_CONFIRMATION_POLL_INTERVAL = 4

Default interval (in seconds) between polling attempts for transaction confirmation.

ESCROW_BULK_PAYOUT_MAX_ITEMS `module-attribute` ¶

ESCROW_BULK_PAYOUT_MAX_ITEMS = 99

Maximum number of recipients allowed in a single bulk payout transaction.

ChainId ¶

Bases: Enum

Supported blockchain network identifiers.

Attributes:

Name	Type	Description
`MAINNET`		Ethereum Mainnet (Chain ID: 1)
`SEPOLIA`		Ethereum Sepolia Testnet (Chain ID: 11155111)
`BSC_MAINNET`		Binance Smart Chain Mainnet (Chain ID: 56)
`BSC_TESTNET`		Binance Smart Chain Testnet (Chain ID: 97)
`POLYGON`		Polygon Mainnet (Chain ID: 137)
`POLYGON_AMOY`		Polygon Amoy Testnet (Chain ID: 80002)
`LOCALHOST`		Local development network (Chain ID: 1338)

KVStoreKeys ¶

Bases: Enum

Standard key names for the on-chain key-value store.

These keys are used by operators to store configuration and metadata on-chain.

Attributes:

Name	Type	Description
`category`		Operator category classification.
`fee`		Operator fee percentage.
`job_types`		Comma-separated list of supported job types.
`operator_name`		Display name of the operator.
`public_key`		PGP public key for encrypted communication.
`public_key_hash`		Hash of the public key file.
`registration_instructions`		Instructions for worker registration.
`registration_needed`		Whether registration is required (boolean).
`role`		Operator role identifier.
`url`		Primary URL for the operator.
`website`		Public-facing website URL.
`webhook_url`		Webhook endpoint for notifications.

OperatorCategory ¶

Bases: Enum

Categories for operator classification.

Attributes:

Name	Type	Description
`MACHINE_LEARNING`		Operators providing machine learning services.
`MARKET_MAKING`		Operators providing market making services.

OrderDirection ¶

Bases: Enum

Sort order for query results.

Attributes:

Name	Type	Description
`ASC`		Ascending order (lowest to highest).
`DESC`		Descending order (highest to lowest).

Role ¶

Bases: Enum

Oracle and operator role identifiers.

Attributes:

Name	Type	Description
`job_launcher`		Entity that creates and funds escrows.
`exchange_oracle`		Oracle handling job distribution and exchange.
`reputation_oracle`		Oracle managing worker reputation scoring.
`recording_oracle`		Oracle recording and validating job results.

Status ¶

Bases: Enum

Escrow contract lifecycle statuses.

Attributes:

Name	Type	Description
`Launched`		Escrow created but not yet funded or configured.
`Pending`		Escrow funded and awaiting oracle actions.
`Partial`		Escrow partially paid out to workers.
`Paid`		All funds distributed but not yet marked complete.
`Complete`		Escrow fully processed and finalized.
`Cancelled`		Escrow cancelled and funds refunded.
`ToCancel`		Cancellation requested, awaiting finalization.

Filters¶

Filter classes for querying Human Protocol subgraph data.

This module provides filter classes for various query operations including escrows, payouts, transactions, statistics, and more.

CancellationRefundFilter ¶

CancellationRefundFilter(
    chain_id,
    escrow_address=None,
    receiver=None,
    date_from=None,
    date_to=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.DESC,
)

Filter configuration for querying cancellation refund events.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain to request refund data from.
`escrow_address`	`Optional[str]`	Escrow address to filter by.
`receiver`	`Optional[str]`	Receiver address to filter by.
`date_from`	`Optional[datetime]`	Filter refunds from this date.
`date_to`	`Optional[datetime]`	Filter refunds until this date.
`first`	`int`	Number of items per page.
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Raises:

Type	Description
`FilterError`	If chain ID is invalid, addresses are malformed, or date range is invalid.

EscrowFilter ¶

EscrowFilter(
    chain_id,
    launcher=None,
    reputation_oracle=None,
    recording_oracle=None,
    exchange_oracle=None,
    job_requester_id=None,
    status=None,
    date_from=None,
    date_to=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.DESC,
)

Filter configuration for querying escrows from the subgraph.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Network to request data from.
`launcher`	`Optional[str]`	Launcher address to filter by.
`reputation_oracle`	`Optional[str]`	Reputation oracle address to filter by.
`recording_oracle`	`Optional[str]`	Recording oracle address to filter by.
`exchange_oracle`	`Optional[str]`	Exchange oracle address to filter by.
`job_requester_id`	`Optional[str]`	Job requester identifier to filter by.
`status`	`Optional[Status \| List[Status]]`	Escrow status or list of statuses to filter by.
`date_from`	`Optional[datetime]`	Filter escrows created from this date.
`date_to`	`Optional[datetime]`	Filter escrows created until this date.
`first`	`int`	Number of items per page (max 1000).
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Raises:

Type	Description
`FilterError`	If chain ID is invalid, addresses are malformed, date range is invalid, or order direction is invalid.

FilterError ¶

Bases: Exception

Exception raised when filter construction or validation fails.

PayoutFilter ¶

PayoutFilter(
    chain_id,
    escrow_address=None,
    recipient=None,
    date_from=None,
    date_to=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.DESC,
)

Filter configuration for querying payout events from the subgraph.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain where payouts were recorded.
`escrow_address`	`Optional[str]`	Escrow address to filter payouts by.
`recipient`	`Optional[str]`	Recipient address to filter payouts by.
`date_from`	`Optional[datetime]`	Filter payouts from this date.
`date_to`	`Optional[datetime]`	Filter payouts until this date.
`first`	`int`	Number of items per page.
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Raises:

Type	Description
`FilterError`	If addresses are malformed or date range is invalid.

StakersFilter ¶

StakersFilter(
    chain_id,
    min_staked_amount=None,
    max_staked_amount=None,
    min_locked_amount=None,
    max_locked_amount=None,
    min_withdrawn_amount=None,
    max_withdrawn_amount=None,
    min_slashed_amount=None,
    max_slashed_amount=None,
    order_by="lastDepositTimestamp",
    order_direction=OrderDirection.DESC,
    first=10,
    skip=0,
)

Filter configuration for querying staker data from the subgraph.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain to request staker data from.
`min_staked_amount`	`Optional[str]`	Minimum staked amount to filter by.
`max_staked_amount`	`Optional[str]`	Maximum staked amount to filter by.
`min_locked_amount`	`Optional[str]`	Minimum locked amount to filter by.
`max_locked_amount`	`Optional[str]`	Maximum locked amount to filter by.
`min_withdrawn_amount`	`Optional[str]`	Minimum withdrawn amount to filter by.
`max_withdrawn_amount`	`Optional[str]`	Maximum withdrawn amount to filter by.
`min_slashed_amount`	`Optional[str]`	Minimum slashed amount to filter by.
`max_slashed_amount`	`Optional[str]`	Maximum slashed amount to filter by.
`order_by`	`Optional[str]`	Property to order results by (e.g., "lastDepositTimestamp").
`order_direction`	`OrderDirection`	Sort order for results.
`first`	`Optional[int]`	Number of items per page.
`skip`	`Optional[int]`	Number of items to skip for pagination.

StatisticsFilter ¶

StatisticsFilter(
    date_from=None,
    date_to=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.ASC,
)

Filter configuration for querying statistical data from the subgraph.

Attributes:

Name	Type	Description
`date_from`	`Optional[datetime]`	Start date for the query range.
`date_to`	`Optional[datetime]`	End date for the query range.
`first`	`int`	Number of items per page (max 1000).
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Example

from datetime import datetime
from human_protocol_sdk.filter import StatisticsFilter
from human_protocol_sdk.constants import OrderDirection

filter = StatisticsFilter(
    date_from=datetime(2023, 1, 1),
    date_to=datetime(2023, 12, 31),
    first=10,
    skip=0,
    order_direction=OrderDirection.ASC
)

Raises:

Type	Description
`FilterError`	If date range is invalid or order direction is invalid.

StatusEventFilter ¶

StatusEventFilter(
    chain_id,
    statuses=None,
    date_from=None,
    date_to=None,
    launcher=None,
    escrow_address=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.DESC,
)

Filter configuration for querying escrow status change events.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain where status events were recorded.
`statuses`	`List[Status]`	List of statuses to filter by.
`date_from`	`Optional[datetime]`	Filter events from this date.
`date_to`	`Optional[datetime]`	Filter events until this date.
`launcher`	`Optional[str]`	Launcher address to filter by.
`escrow_address`	`Optional[str]`	Escrow address to filter by.
`first`	`int`	Number of items per page.
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Initializes a filter for status events.

:param chain_id: The chain ID where the events are recorded. :param statuses: Optional list of statuses to filter by. :param date_from: Optional start date for filtering. :param date_to: Optional end date for filtering. :param launcher: Optional launcher address to filter by. :param escrow_address: Optional escrow address to filter by. :param first: Optional number of events per page. Default is 10. :param skip: Optional number of events to skip. Default is 0. :param order_direction: Optional order direction. Default is DESC.

TransactionFilter ¶

TransactionFilter(
    chain_id,
    from_address=None,
    to_address=None,
    start_date=None,
    end_date=None,
    start_block=None,
    end_block=None,
    method=None,
    escrow=None,
    token=None,
    first=10,
    skip=0,
    order_direction=OrderDirection.DESC,
)

Filter configuration for querying blockchain transactions from the subgraph.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain to filter transactions from.
`from_address`	`Optional[str]`	Sender address to filter by.
`to_address`	`Optional[str]`	Recipient address to filter by.
`start_date`	`Optional[datetime]`	Filter transactions from this date.
`end_date`	`Optional[datetime]`	Filter transactions until this date.
`start_block`	`Optional[int]`	Filter transactions from this block number.
`end_block`	`Optional[int]`	Filter transactions until this block number.
`method`	`Optional[str]`	Method signature to filter transactions by.
`escrow`	`Optional[str]`	Escrow address to filter transactions by.
`token`	`Optional[str]`	Token address to filter transactions by.
`first`	`int`	Number of items per page (max 1000).
`skip`	`int`	Number of items to skip for pagination.
`order_direction`	`OrderDirection`	Sort order for results.

Raises:

Type	Description
`ValueError`	If addresses are malformed, date/block ranges are invalid, or order direction is invalid.

WorkerFilter ¶

WorkerFilter(
    chain_id,
    worker_address=None,
    order_by="payoutCount",
    order_direction=OrderDirection.DESC,
    first=10,
    skip=0,
)

Filter configuration for querying worker data from the subgraph.

Attributes:

Name	Type	Description
`chain_id`	`ChainId`	Chain to request worker data from.
`worker_address`	`Optional[str]`	Worker address to filter by.
`order_by`	`Optional[str]`	Property to order results by (e.g., "payoutCount").
`order_direction`	`OrderDirection`	Sort order for results.
`first`	`int`	Number of items per page (1-1000).
`skip`	`int`	Number of items to skip for pagination.

Raises:

Type	Description
`FilterError`	If order direction is invalid.

Decorators¶

Decorators for SDK functionality.

RequiresSignerError ¶

Bases: Exception

Raised when a transaction-signing method is called without proper Web3 account configuration.

This exception is raised by the @requires_signer decorator when a method requiring transaction signing capabilities is invoked on a Web3 instance that lacks:

A default account (w3.eth.default_account)
SignAndSendRawMiddlewareBuilder middleware for transaction signing

requires_signer ¶

requires_signer(method)

Decorator that ensures Web3 instance has signing capabilities before executing a method.

This decorator validates that the Web3 instance has both a default account configured and the SignAndSendRawMiddlewareBuilder middleware installed. These are required for methods that need to sign and send transactions.

Parameters:

Name	Type	Description	Default
`method`	`Callable[..., Any]`	The method to decorate (must be a method of a class with a `w3` attribute).	required

Returns:

Type	Description
`Callable[..., Any]`	Wrapped method that performs validation before execution.

Raises:

Type	Description
`RequiresSignerError`	If the Web3 instance lacks a default account or signing middleware.

Example

from web3 import Web3
from web3.middleware import SignAndSendRawMiddlewareBuilder
from human_protocol_sdk.decorators import requires_signer

class MyClient:
    def __init__(self, w3):
        self.w3 = w3

    @requires_signer
    def send_transaction(self):
        # This method requires a signer
        pass

General utilities¶

SubgraphOptions `dataclass` ¶

SubgraphOptions(
    max_retries=None, base_delay=None, indexer_id=None
)

Configuration options for subgraph queries with retry logic and indexer routing.

Attributes:

Name	Type	Description
`max_retries`	`Optional[int]`	Maximum number of retry attempts for failed queries. Must be paired with base_delay.
`base_delay`	`Optional[int]`	Base delay in milliseconds between retry attempts. Must be paired with max_retries.
`indexer_id`	`Optional[str]`	Specific indexer ID to route requests to (requires SUBGRAPH_API_KEY environment variable).

TransactionOptions ¶

Bases: TxParams

Transaction params extended with wait configuration hints.

WaitOptions `dataclass` ¶

WaitOptions(confirmations=None, timeout_ms=None)

Configuration for transaction receipt waiting logic.

custom_gql_fetch ¶

custom_gql_fetch(
    network,
    query,
    params=None,
    options=None,
    use_hmt_subgraph=False,
)

Fetch data from the subgraph with optional retry logic and indexer routing.

Parameters:

Name	Type	Description	Default
`network`	`Dict[str, Any]`	Network configuration dictionary containing subgraph URLs.	required
`query`	`str`	GraphQL query string to execute.	required
`params`	`Optional[Dict[str, Any]]`	Optional query parameters/variables dictionary.	`None`
`options`	`Optional[SubgraphOptions]`	Optional subgraph configuration for retries and indexer selection.	`None`
`use_hmt_subgraph`	`bool`	If true, resolves URL using HMT subgraph keys with fallback.	`False`

Returns:

Type	Description
`Dict[str, Any]`	JSON response from the subgraph containing the query results.

Raises:

Type	Description
`ValueError`	If retry configuration is incomplete or indexer routing requires missing API key.
`Exception`	If the subgraph query fails after all retry attempts.

Example

from human_protocol_sdk.constants import NETWORKS, ChainId
from human_protocol_sdk.utils import SubgraphOptions, custom_gql_fetch

network = NETWORKS[ChainId.POLYGON_AMOY]
query = "{ escrows(first: 10) { id address } }"

# Simple query
data = custom_gql_fetch(network, query)

# With retry logic
data = custom_gql_fetch(
    network,
    query,
    options=SubgraphOptions(max_retries=3, base_delay=1000)
)

get_contract_interface ¶

get_contract_interface(contract_entrypoint)

Retrieve the contract ABI and interface from a compiled artifact file.

Parameters:

Name	Type	Description	Default
`contract_entrypoint`	`str`	File path to the contract JSON artifact.	required

Returns:

Type	Description
`Dict[str, Any]`	Contract interface dictionary containing the ABI and other metadata.

Example

interface = get_contract_interface("artifacts/contracts/MyContract.sol/MyContract.json")
abi = interface["abi"]

get_erc20_interface ¶

get_erc20_interface()

Retrieve the standard ERC20 token contract interface.

Returns:

Type	Description
`Dict[str, Any]`	The ERC20 contract interface containing the ABI.

Example

erc20_interface = get_erc20_interface()
token_contract = w3.eth.contract(address=token_address, abi=erc20_interface["abi"])

get_error_message ¶

get_error_message(error)

Extract a readable error message from an exception-like object.

get_escrow_interface ¶

get_escrow_interface()

Retrieve the Escrow contract interface.

Returns:

Type	Description
`Dict[str, Any]`	The Escrow contract interface containing the ABI.

Example

escrow_interface = get_escrow_interface()
escrow_contract = w3.eth.contract(address=escrow_address, abi=escrow_interface["abi"])

get_factory_interface ¶

get_factory_interface()

Retrieve the EscrowFactory contract interface.

Returns:

Type	Description
`Dict[str, Any]`	The EscrowFactory contract interface containing the ABI.

Example

factory_interface = get_factory_interface()
factory_contract = w3.eth.contract(address=factory_address, abi=factory_interface["abi"])

get_hmt_balance ¶

get_hmt_balance(wallet_addr, token_addr, w3)

Get the HMT token balance for a wallet address.

Parameters:

Name	Type	Description	Default
`wallet_addr`	`str`	Wallet address to check balance for.	required
`token_addr`	`str`	ERC-20 token contract address.	required
`w3`	`Web3`	Web3 instance connected to the network.	required

Returns:

Type	Description
`int`	HMT token balance in wei.

Example

from web3 import Web3

w3 = Web3(Web3.HTTPProvider("https://polygon-rpc.com"))
balance = get_hmt_balance(
    "0x1234567890123456789012345678901234567890",
    "0xc748B2A084F8eFc47E086ccdDD9b7e67aEb571BF",
    w3
)

get_kvstore_interface ¶

get_kvstore_interface()

Retrieve the KVStore contract interface.

Returns:

Type	Description
`Dict[str, Any]`	The KVStore contract interface containing the ABI.

Example

kvstore_interface = get_kvstore_interface()
kvstore_contract = w3.eth.contract(address=kvstore_address, abi=kvstore_interface["abi"])

get_staking_interface ¶

get_staking_interface()

Retrieve the Staking contract interface.

Returns:

Type	Description
`Dict[str, Any]`	The Staking contract interface containing the ABI.

Example

staking_interface = get_staking_interface()
staking_contract = w3.eth.contract(address=staking_address, abi=staking_interface["abi"])

handle_error ¶

handle_error(e, exception_class)

Handle and translate errors raised during contract transactions.

This function captures exceptions (especially ContractLogicError from web3.py), extracts meaningful revert reasons when present, logs unexpected errors, and raises a custom exception with a clear message for SDK users.

Parameters:

Name	Type	Description	Default
`e`	`Exception`	The exception object raised during a transaction.	required
`exception_class`	`Type[Exception]`	The custom exception class to raise (e.g., EscrowClientError).	required

Raises:

Type	Description
`exception_class`	Always raises the provided exception class with a formatted error message.

Example

from human_protocol_sdk.escrow import EscrowClientError

try:
    tx_hash = contract.functions.someMethod(...).transact()
    w3.eth.wait_for_transaction_receipt(tx_hash)
except Exception as e:
    handle_error(e, EscrowClientError)

is_indexer_error ¶

is_indexer_error(error)

Check if an error indicates that The Graph indexer is down or not synced.

This function inspects error responses from The Graph API to detect "bad indexers" messages that indicate infrastructure issues rather than query problems.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception to check.	required

Returns:

Type	Description
`bool`	True if the error indicates indexer issues, False otherwise.

Example

try:
    data = custom_gql_fetch(network, query)
except Exception as e:
    if is_indexer_error(e):
        # Retry with different indexer
        pass

normalize_wait_tx_options ¶

normalize_wait_tx_options(tx_options, error_class)

Split tx options from wait parameters while validating their values.

parse_transfer_transaction ¶

parse_transfer_transaction(hmtoken_contract, tx_receipt)

Parse a transaction receipt to extract HMT transfer information.

Parameters:

Name	Type	Description	Default
`hmtoken_contract`	`Contract`	The HMT token contract instance.	required
`tx_receipt`	`Optional[TxReceipt]`	Transaction receipt to parse, or None.	required

Returns:

Type	Description
`Tuple[bool, Optional[int]]`	A tuple containing: bool: True if HMT was successfully transferred, False otherwise. Optional[int]: The transfer amount in wei if successful, None otherwise.

Example

from web3 import Web3

tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
transferred, amount = parse_transfer_transaction(hmt_contract, tx_receipt)
if transferred:
    print(f"Transferred {amount} wei")

transact_and_wait ¶

transact_and_wait(
    w3, contract_function, tx_options, error_class
)

Send a contract transaction and wait for confirmations.

validate_json ¶

validate_json(data)

Validate whether a string contains valid JSON data.

Parameters:

Name	Type	Description	Default
`data`	`str`	String to validate as JSON.	required

Returns:

Type	Description
`bool`	True if the string is valid JSON, False otherwise.

Example

from human_protocol_sdk.utils import validate_json

if validate_json('{"key": "value"}'):
    print("Valid JSON")

validate_url ¶

validate_url(url)

Validate whether a string is a properly formatted URL.

This function supports both standard URLs and Docker network URLs that may not be recognized by strict validators.

Parameters:

Name	Type	Description	Default
`url`	`str`	URL string to validate (e.g., "https://example.com" or "http://localhost:8080").	required

Returns:

Type	Description
`bool`	True if the URL is valid, False otherwise.

Raises:

Type	Description
`ValidationFailure`	If the URL format is invalid according to the validators library.

Example

from human_protocol_sdk.utils import validate_url

if validate_url("https://example.com"):
    print("Valid URL")

wait_for_transaction_receipt_with_confirmations ¶

wait_for_transaction_receipt_with_confirmations(
    w3,
    tx_hash,
    wait_options,
    error_class,
    poll_interval=DEFAULT_CONFIRMATION_POLL_INTERVAL,
)

Waits for a receipt and enforces optional confirmations & timeout.

Core utilities¶

Constants¶

ARTIFACTS_FOLDER module-attribute ¶

DEFAULT_CONFIRMATION_POLL_INTERVAL module-attribute ¶

ESCROW_BULK_PAYOUT_MAX_ITEMS module-attribute ¶

ChainId ¶

KVStoreKeys ¶

OperatorCategory ¶

OrderDirection ¶

Role ¶

Status ¶

Filters¶

CancellationRefundFilter ¶

EscrowFilter ¶

FilterError ¶

PayoutFilter ¶

StakersFilter ¶

StatisticsFilter ¶

StatusEventFilter ¶

TransactionFilter ¶

WorkerFilter ¶

Decorators¶

RequiresSignerError ¶

requires_signer ¶

General utilities¶

SubgraphOptions dataclass ¶

TransactionOptions ¶

WaitOptions dataclass ¶

custom_gql_fetch ¶

get_contract_interface ¶

get_erc20_interface ¶

get_error_message ¶

get_escrow_interface ¶

get_factory_interface ¶

get_hmt_balance ¶

get_kvstore_interface ¶

get_staking_interface ¶

handle_error ¶

is_indexer_error ¶

normalize_wait_tx_options ¶

parse_transfer_transaction ¶

transact_and_wait ¶

validate_json ¶

validate_url ¶

wait_for_transaction_receipt_with_confirmations ¶

ARTIFACTS_FOLDER `module-attribute` ¶

DEFAULT_CONFIRMATION_POLL_INTERVAL `module-attribute` ¶

ESCROW_BULK_PAYOUT_MAX_ITEMS `module-attribute` ¶

SubgraphOptions `dataclass` ¶

WaitOptions `dataclass` ¶